Utils

class todd.utils.BaseHandler[source]

Bases: Generic[T], ABC

Utility class for working with collections.

abstractmethod classmethod can_handle(obj)[source]

Parameters:: obj (Any)
Return type:: TypeGuard[T]

abstractmethod classmethod elements(obj)[source]

Parameters:: obj (T)
Return type:: list[Any]

abstractmethod classmethod map(f, *objs)[source]

Apply a function to the given objects.

All the objects must be of the same shape.

Parameters:

f (CallableProtocol[Any, Any])
objs (T)

Return type:

T

class todd.utils.CallableProtocol[source]

Bases: Protocol[T_contra, T_co]

__init__(*args, **kwargs)

class todd.utils.EMA[source]

Bases: object

__init__(decay=0.99)[source]

Parameters:: decay (Any)
Return type:: None

classmethod check_decay(decay)[source]

Parameters:: decay (Any)
Return type:: None

property decay: Any

class todd.utils.EnvRegistry[source]

Bases: Registry

data = {'cuda_home': <function cuda_home>, 'git_commit_id': <function git_commit_id>, 'git_status': <function git_status>, 'nvidia_smi': <function nvidia_smi>, 'opencv_version': <function opencv_version>, 'platform': <function platform>, 'python_version': <function python_version>, 'pytorch_version': <function pytorch_version>, 'todd_version': <function todd_version>, 'torchvision_version': <function torchvision_version>}

class todd.utils.HandlerRegistry[source]

Bases: Registry

data = {'MappingHandler': <class 'todd.utils.nested_collection_utils.MappingHandler'>, 'SequenceHandler': <class 'todd.utils.nested_collection_utils.SequenceHandler'>, 'SetHandler': <class 'todd.utils.nested_collection_utils.SetHandler'>}

class todd.utils.HolderMixin[source]

Bases: Generic[T]

__init__(*args, instance=None, **kwargs)[source]

Parameters:: instance (T | None)
Return type:: None

bind(instance)[source]

Parameters:: instance (T)
Return type:: None

property holding: bool

class todd.utils.Keys[source]

Bases: NamedTuple

Keys(missing, unexpected)

static __new__(_cls, missing, unexpected)

Create new instance of Keys(missing, unexpected)

Parameters:

missing (list[str])
unexpected (list[str])

missing: list[str]: Alias for field number 0

unexpected: list[str]: Alias for field number 1

class todd.utils.MappingHandler[source]

Bases: BaseHandler[Mapping[KT, VT]]

classmethod can_handle(obj)[source]

Parameters:: obj (Any)
Return type:: TypeGuard[Mapping[KT, VT]]

classmethod elements(obj)[source]

Parameters:: obj (Mapping[KT, VT])
Return type:: list[VT]

classmethod map(f, *objs)[source]

Parameters:

f (CallableProtocol[VT, T_co])
objs (Mapping[KT, VT])

Return type:

dict[KT, T_co]

class todd.utils.NestedCollectionUtils[source]

Bases: object

__init__(*args, atomic_types=None, **kwargs)[source]

Parameters:: atomic_types (Iterable[type[Any]] | None)
Return type:: None

add_atomic_type(*types)[source]

Parameters:: types (type[Any])
Return type:: None

can_handle(obj)[source]

Parameters:: obj (Any)
Return type:: bool

elements(obj)[source]

Get the elements of the given object.

Parameters:: obj (Any) – The object to get the elements from.
Returns:: Elements of the given object.
Return type:: list[Any]

Examples

>>> utils = NestedCollectionUtils()
>>> list(utils.elements([]))
[]
>>> list(utils.elements({1: 'a', 2: 'b'}))
['a', 'b']
>>> list(utils.elements({1, 2, 3}))
[1, 2, 3]
>>> list(utils.elements(('a', 'b', 'c')))
['a', 'b', 'c']
>>> utils.elements([[1, 2], [3, [4, 5]]])
[1, 2, 3, 4, 5]
>>> utils.elements([1, {2: 'a'}, (3, 4)])
[1, 'a', 3, 4]

get_handler(*objs)[source]

Find a utility class for all the given object.

Parameters:: objs (Any) – The objects to check.
Returns:: The utility class or None if none of the utility classes is applicable.
Return type:: type[BaseHandler[Any]] | None

Examples

>>> utils = NestedCollectionUtils()
>>> utils.get_handler([])
<class '...SequenceHandler'>
>>> utils.get_handler(tuple())
<class '...SequenceHandler'>
>>> utils.get_handler(dict())
<class '...MappingHandler'>
>>> utils.get_handler(set())
<class '...SetHandler'>
>>> utils.get_handler('')
>>> utils.get_handler(123)
>>> utils.get_handler(None)

index(obj, indices)[source]

Index the given object with the given indices.

Parameters:

obj (Any) – The object to index.
indices (Any) – The indices to use for indexing.

Returns:

The result of indexing the object with the indices.

Return type:

Any

Examples

>>> utils = NestedCollectionUtils()
>>> utils.index([1, 2, 3], 0)
1
>>> utils.index({1: 'a', 2: 'b', 3: 'c'}, 1)
'a'
>>> utils.index([[1, 2], [3, 4], [5, 6]], [0, 1])
2

is_atomic(obj)[source]

Parameters:: obj (Any)
Return type:: bool

is_atomic_collection(obj)[source]

Check if the given object is atomic.

An object is considered atomic if all its elements are not collections.

Parameters:: obj (Any) – The object to check.
Returns:: True if the object is atomic, False otherwise.
Return type:: bool

Examples

>>> utils = NestedCollectionUtils()
>>> utils.is_atomic_collection([])
True
>>> utils.is_atomic_collection({1: 'a', 2: 'b'})
True
>>> utils.is_atomic_collection({1, 2, 3})
True
>>> utils.is_atomic_collection(('a', 'b', 'c'))
True
>>> utils.is_atomic_collection([1, [2, 3], [4, [5, 6]]])
False
>>> utils.is_atomic_collection({1: [2, 3], 4: [5, 6]})
False

map(f, *objs)[source]

Recursively apply a function to the given objects.

Parameters:

f (CallableProtocol[Any, Any]) – The function to apply.
objs (Any) – The objects to apply the function to.

Returns:

The result of applying the function to the objects.

Return type:

Any

Examples

>>> utils = NestedCollectionUtils()
>>> utils.map(lambda x: x + 1, [1, 2, 3])
(2, 3, 4)
>>> result = utils.map(
...     lambda x: x.upper(),
...     {'a': 'apple', 'b': 'banana'},
... )
>>> dict(sorted(result.items()))
{'a': 'APPLE', 'b': 'BANANA'}
>>> utils.map(lambda x: x * 2, set([1, 2, 3]))
{2, 4, 6}
>>> utils.map(lambda x: x.lower(), ('HELLO', 'WORLD'))
('hello', 'world')

reduce(f, obj)[source]

Apply a function to the collection and returns a single value.

Parameters:

f (Callable[[Iterable[Any]], Any]) – The function to apply to the elements.
obj (Any) – The collection to reduce.

Returns:

The result of applying the function to the collection.

Return type:

Any

Examples

>>> utils = NestedCollectionUtils()
>>> utils.reduce(sum, [])
0
>>> utils.reduce(sum, [1])
1
>>> utils.reduce(sum, [1, 2, 3, 4])
10
>>> utils.reduce(sum, [[1, 2], [3, 4], [5, 6]])
21

class todd.utils.NestedTensorCollectionUtils[source]

Bases: NestedCollectionUtils

all_close(x, y, **kwargs)[source]

Parameters:

x (Any)
y (Any)

Return type:

bool

index(obj, indices)[source]

Parameters:

obj (Any)
indices (Tensor)

Return type:

Tensor

new_empty(obj, *args, **kwargs)[source]

Parameters:: obj (Any)
Return type:: Tensor

shape(obj, depth=0)[source]

Parameters:

obj (Any)
depth (int)

Return type:

tuple[int, …]

stack(obj, **kwargs)[source]

Parameters:: obj (Any)
Return type:: Tensor

class todd.utils.SequenceHandler[source]

Bases: BaseHandler[Sequence[T]]

classmethod can_handle(obj)[source]

Parameters:: obj (Any)
Return type:: TypeGuard[Sequence[T]]

classmethod elements(obj)[source]

Parameters:: obj (Sequence[T])
Return type:: list[T]

classmethod map(f, *objs)[source]

Parameters:

f (CallableProtocol[T, T_co])
objs (Sequence[T])

Return type:

tuple[T_co, …]

class todd.utils.SerializeMixin[source]: Bases: ABC

class todd.utils.SetHandler[source]

Bases: BaseHandler[Set[T]]

classmethod can_handle(obj)[source]

Parameters:: obj (Any)
Return type:: TypeGuard[Set[T]]

classmethod elements(obj)[source]

Parameters:: obj (Set[T])
Return type:: list[T]

classmethod map(f, *objs)[source]

Parameters:

f (CallableProtocol[T, T_co])
objs (Set[T])

Return type:

set[T_co]

class todd.utils.StateDictConverter[source]

Bases: object

__init__(*args, module=None, **kwargs)[source]

Parameters:: module (Module | None)
Return type:: None

convert(state_dict)[source]

Parameters:: state_dict (dict[str, Tensor])
Return type:: dict[str, Tensor]

load(*args, **kwargs)[source]

Return type:: dict[str, Tensor]

property module: Module

class todd.utils.StateDictMixin[source]

Bases: object

load_state_dict(state_dict, *args, **kwargs)[source]

Parameters:: state_dict (Mapping[str, Any])
Return type:: Keys | None

state_dict(*args, **kwargs)[source]

Return type:: dict[str, Any]

class todd.utils.StateMachine[source]

Bases: Generic[T]

A class to represent a state that can be transited to another state.

Examples

>>> class Enum(enum.Enum):
...     A = 1
...     B = 2
...     C = 3
>>> state = StateMachine(Enum.A)
>>> state.state
<Enum.A: 1>
>>> state.transit({Enum.A: Enum.B})
>>> state.state
<Enum.B: 2>
>>> state.transit({Enum.A: Enum.C})
Traceback (most recent call last):
    ...
RuntimeError: Enum.B is not in {<Enum.A: 1>: <Enum.C: 3>}.

__init__(state)[source]

Parameters:: state (T)
Return type:: None

property state: T

transit(transitions)[source]

Parameters:: transitions (Mapping[T, T])
Return type:: None

class todd.utils.Statistician[source]

Bases: object

__init__(chunk_size)[source]

Parameters:: chunk_size (int)
Return type:: None

compute_mean()[source]

Return type:: Tensor

compute_variance(mean=None)[source]

Parameters:: mean (Tensor | None)
Return type:: Tensor

property num_samples: int

update(samples)[source]

Parameters:: samples (Tensor)
Return type:: None

class todd.utils.Store[source]

Bases: object

DEVICE: str = 'cpu'

DRY_RUN: bool = False

TRAIN_WITH_VAL_DATASET: bool = False

cpu = True

cuda = False

mps = False

class todd.utils.StoreMeta[source]

Bases: NonInstantiableMeta

Stores for global variables.

Stores provide an interface to access global variables:

>>> class CustomStore(metaclass=StoreMeta):
...     VARIABLE: int
>>> CustomStore.VARIABLE
0
>>> CustomStore.VARIABLE = 1
>>> CustomStore.VARIABLE
1

Variables can have explicit default values:

>>> class DefaultStore(metaclass=StoreMeta):
...     DEFAULT: float = 0.625
>>> DefaultStore.DEFAULT
0.625

Non-empty environment variables are read-only. For string variables, their values are read directly from the environment. Other environment variables are evaluated and should be of the corresponding type. Default values are ignored.

>>> os.environ['ENV_INT'] = '2'
>>> os.environ['ENV_STR'] = 'hello world!'
>>> os.environ['ENV_DICT'] = 'dict(a=1)'
>>> class EnvStore(metaclass=StoreMeta):
...     ENV_INT: int = 1
...     ENV_STR: str
...     ENV_DICT: dict
>>> EnvStore.ENV_INT
2
>>> EnvStore.ENV_STR
'hello world!'
>>> EnvStore.ENV_DICT
{'a': 1}

Assignments to those variables will not trigger exceptions, but will not take effect:

>>> EnvStore.ENV_INT = 3
>>> EnvStore.ENV_INT
2

__init__(*args, **kwargs)[source]

Return type:: None

todd.utils.collect_env_(*args, **kwargs)[source]

Return type:: str

todd.utils.cuda_home(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.ema(x, y, decay)[source]

Parameters:

x (Any)
y (Any)
decay (Any)

Return type:

Any

todd.utils.get_audio(url)[source]

Parameters:: url (str)
Return type:: tuple[Tensor, int]

todd.utils.get_bytes(url)[source]

Parameters:: url (str)
Return type:: BytesIO

todd.utils.get_image(url)[source]

Parameters:: url (str)
Return type:: ndarray[tuple[Any, …], dtype[uint8]]

todd.utils.get_timestamp()[source]

Return type:: str

todd.utils.git_commit_id(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.git_status(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.init_seed(seed)[source]

Parameters:: seed (int)
Return type:: None

todd.utils.is_sync(x)[source]

Parameters:: x (Tensor)
Return type:: bool

todd.utils.nvidia_smi(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.opencv_version(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.platform(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.python_version(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.pytorch_version(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.retry(n)[source]

Parameters:: n (int)
Return type:: Any

todd.utils.set_seed_temp(seed=None, deterministic=False, benchmark=True)[source]

Parameters:

deterministic (bool)
benchmark (bool)

Return type:

Generator[None, None, None]

todd.utils.set_temp(obj, name, value)[source]

Set a temporary attribute on an object.

Parameters:

obj – The object to set the attribute on.
name (str) – The attribute name.
value – The value to set.

Return type:

Generator[None, None, None]

todd.utils.todd_version(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.torchvision_version(verbose=False)[source]

Parameters:: verbose (bool)
Return type:: str | None

todd.utils.transfer_state_dict(target, source)[source]

Parameters:

target (Module | StateDictMixin)
source (Module | StateDictMixin)

Return type:

None

todd.utils.transfer_state_dicts(models, prefixes)[source]

Parameters:

models (Any)
prefixes (Mapping[str, str])

Return type:

None