Utilities¶

Main
- Config
- Data
- Distributed
- Misc
- Numpy
- Onnx
- Pruning
- Quantization
- Stochastic Weights Averaging (SWA)
- Sys
- Torch
- Tracing
Contrib
- Compression
- Image
- Pandas
- Parallel
- Serialization
- Visualization

Main ¶

Config ¶

catalyst.utils.config.load_config(path: Union[str, pathlib.Path], ordered: bool = False, data_format: str = None, encoding: str = 'utf-8') → Union[Dict, List][source]¶

Loads config by giving path. Supports YAML and JSON files.

Examples

>>> load(path="./config.yml", ordered=True)

Parameters

path – path to config file (YAML or JSON)
ordered – if true the config will be loaded as OrderedDict
data_format – yaml, yml or json.
encoding – encoding to read the config

Returns

config

Return type

Union[Dict, List]

Raises

ValueError – if path path doesn’t exists or file format is not YAML or JSON

Adapted from https://github.com/TezRomacH/safitty/blob/v1.2.0/safitty/parser.py#L63 which was adapted from https://github.com/catalyst-team/catalyst/blob/v19.03/catalyst/utils/config.py#L10

catalyst.utils.config.save_config(config: Union[Dict, List], path: Union[str, pathlib.Path], data_format: str = None, encoding: str = 'utf-8', ensure_ascii: bool = False, indent: int = 2) → None[source]¶

Saves config to file. Path must be either YAML or JSON.

Parameters

config (Union[Dict, List]) – config to save
path (Union[str, Path]) – path to save
data_format – yaml, yml or json.
encoding – Encoding to write file. Default is utf-8
ensure_ascii – Used for JSON, if True non-ASCII
are escaped in JSON strings. (characters) –
indent – Used for JSON

Adapted from https://github.com/TezRomacH/safitty/blob/v1.2.0/safitty/parser.py#L110 which was adapted from https://github.com/catalyst-team/catalyst/blob/v19.03/catalyst/utils/config.py#L38

Data ¶

catalyst.utils.data.get_loader(data_source: Iterable[dict], open_fn: Callable, dict_transform: Callable = None, sampler=None, collate_fn: Callable = <function default_collate>, batch_size: int = 32, num_workers: int = 4, shuffle: bool = False, drop_last: bool = False)[source]¶

Creates a DataLoader from given source and its open/transform params.

Parameters

data_source – and iterable containing your data annotations, (for example path to images, labels, bboxes, etc)
open_fn – function, that can open your annotations dict and transfer it to data, needed by your network (for example open image by path, or tokenize read string)
dict_transform – transforms to use on dict (for example normalize image, add blur, crop/resize/etc)
sampler (Sampler, optional) – defines the strategy to draw samples from the dataset
collate_fn (callable, optional) – merges a list of samples to form a mini-batch of Tensor(s). Used when using batched loading from a map-style dataset
batch_size (int, optional) – how many samples per batch to load
num_workers (int, optional) – how many subprocesses to use for data loading. 0 means that the data will be loaded in the main process
shuffle (bool, optional) – set to True to have the data reshuffled at every epoch (default: False).
drop_last (bool, optional) – set to True to drop the last incomplete batch, if the dataset size is not divisible by the batch size. If False and the size of dataset is not divisible by the batch size, then the last batch will be smaller. (default: False)

Returns

DataLoader with catalyst.data.ListDataset

catalyst.utils.data.get_loaders_from_params(batch_size: int = 1, num_workers: int = 0, drop_last: bool = False, per_gpu_scaling: bool = False, loaders_params: Dict[str, Any] = None, samplers_params: Dict[str, Any] = None, initial_seed: int = 42, datasets_fn: Callable = None, **data_params) → OrderedDict[str, DataLoader][source]¶

Creates pytorch dataloaders from datasets and additional parameters.

Parameters

batch_size – batch_size parameter from torch.utils.data.DataLoader
num_workers – num_workers parameter from torch.utils.data.DataLoader
drop_last – drop_last parameter from torch.utils.data.DataLoader
per_gpu_scaling – boolean flag, if True, scales batch_size in proportion to the number of GPUs
loaders_params (Dict[str, Any]) – additional loaders parameters
samplers_params (Dict[str, Any]) – additional sampler parameters
initial_seed – initial seed for torch.utils.data.DataLoader workers
datasets_fn (Callable) – callable function to get dictionary with torch.utils.data.Datasets
**data_params – additional data parameters or dictionary with torch.utils.data.Datasets to use for pytorch dataloaders creation

Returns

dictionary with: torch.utils.data.DataLoader

Return type

OrderedDict[str, DataLoader]

Raises

NotImplementedError – if datasource is out of Dataset or dict
ValueError – if batch_sampler option is mutually exclusive with distributed

Distributed ¶

catalyst.utils.distributed.get_rank() → int[source]¶

Returns the rank of the current worker.

Returns: rank if torch.distributed is initialized, otherwise -1
Return type: int

catalyst.utils.distributed.get_distributed_params()[source]¶

Returns distributed params for experiment run.

Returns: dictionary with distributed params

catalyst.utils.distributed.get_nn_from_ddp_module(model: torch.nn.modules.module.Module) → torch.nn.modules.module.Module[source]¶

Return a real model from a torch.nn.DataParallel, torch.nn.parallel.DistributedDataParallel, or apex.parallel.DistributedDataParallel.

Parameters: model – A model, or DataParallel wrapper.
Returns: A model

catalyst.utils.distributed.sum_reduce(tensor: torch.Tensor) → torch.Tensor[source]¶

Reduce tensor to all processes and compute total (sum) value.

Parameters: tensor – tensor to reduce.
Returns: reduced tensor

catalyst.utils.distributed.mean_reduce(tensor: torch.Tensor, world_size: int) → torch.Tensor[source]¶

Reduce tensor to all processes and compute mean value.

Parameters

tensor – tensor to reduce.
world_size – number of processes in DDP setup.

Returns

reduced tensor

catalyst.utils.distributed.all_gather(data: Any) → List[Any][source]¶

Run all_gather on arbitrary picklable data (not necessarily tensors).

Note

if data on different devices then data in resulted list will be on the same devices. Source: https://github.com/facebookresearch/detr/blob/master/util/misc.py#L88-L128

Parameters: data – any picklable object
Returns: list of data gathered from each process.

Misc ¶

catalyst.utils.misc.boolean_flag(parser: argparse.ArgumentParser, name: str, default: Optional[bool] = False, help: str = None, shorthand: str = None) → None[source]¶

Add a boolean flag to a parser inplace.

Examples

>>> parser = argparse.ArgumentParser()
>>> boolean_flag(
>>>     parser, "flag", default=False, help="some flag", shorthand="f"
>>> )

Parameters

parser – parser to add the flag to
name – argument name –<name> will enable the flag, while –no-<name> will disable it
default (bool, optional) – default value of the flag
help – help string for the flag
shorthand – shorthand string for the argument

catalyst.utils.misc.get_fn_default_params(fn: Callable[[...], Any], exclude: List[str] = None)[source]¶

Return default parameters of Callable.

Parameters

fn (Callable[.., Any]) – target Callable
exclude – exclude list of parameters

Returns

contains default parameters of fn

Return type

dict

catalyst.utils.misc.get_fn_argsnames(fn: Callable[[...], Any], exclude: List[str] = None)[source]¶

Return parameter names of Callable.

Parameters

fn (Callable[.., Any]) – target Callable
exclude – exclude list of parameters

Returns

contains parameter names of fn

Return type

list

catalyst.utils.misc.get_utcnow_time(format: str = None) → str[source]¶

Return string with current utc time in chosen format.

Parameters: format – format string. if None “%y%m%d.%H%M%S” will be used.
Returns: formatted utc time string
Return type: str

catalyst.utils.misc.is_exception(ex: Any) → bool[source]¶: Check if the argument is of Exception type.

catalyst.utils.misc.maybe_recursive_call(object_or_dict, method: Union[str, Callable], recursive_args=None, recursive_kwargs=None, **kwargs)[source]¶

Calls the method recursively for the object_or_dict.

Parameters

object_or_dict – some object or a dictionary of objects
method – 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

Returns

result of method call

catalyst.utils.misc.get_attr(obj: Any, key: str, inner_key: str = None) → Any[source]¶

Alias for python getattr method. Useful for Callbacks preparation and cases with multi-criterion, multi-optimizer setup. For example, when you would like to train multi-task classification.

Used to get a named attribute from a IRunner by key keyword; for example

get_attr(runner, "criterion")
# is equivalent to
runner.criterion

get_attr(runner, "optimizer")
# is equivalent to
runner.optimizer

get_attr(runner, "scheduler")
# is equivalent to
runner.scheduler

With inner_key usage, it suppose to find a dictionary under key and would get inner_key from this dict; for example,

get_attr(runner, "criterion", "bce")
# is equivalent to
runner.criterion["bce"]

get_attr(runner, "optimizer", "adam")
# is equivalent to
runner.optimizer["adam"]

get_attr(runner, "scheduler", "adam")
# is equivalent to
runner.scheduler["adam"]

Parameters

obj – object of interest
key – name for attribute of interest, like criterion, optimizer, scheduler
inner_key – name of inner dictionary key

Returns

inner attribute

catalyst.utils.misc.set_global_seed(seed: int) → None[source]¶

Sets random seed into Numpy and Random, PyTorch and TensorFlow.

Parameters: seed – random seed

catalyst.utils.misc.get_dictkey_auto_fn(key: Union[str, List[str], None]) → Callable[source]¶

Function generator for sub-dict preparation from dict based on predefined keys.

Parameters: key – keys
Returns: function
Raises: NotImplementedError – if key is out of str, tuple, list, dict, None

catalyst.utils.misc.merge_dicts(*dicts: dict) → dict[source]¶

Recursive dict merge. Instead of updating only top-level keys, merge_dicts recurses down into dicts nested to an arbitrary depth, updating keys.

Parameters: *dicts – several dictionaries to merge
Returns: deep-merged dictionary
Return type: dict

catalyst.utils.misc.flatten_dict(dictionary: Dict[str, Any], parent_key: str = '', separator: str = '/') → collections.OrderedDict[source]¶

Make the given dictionary flatten.

Parameters

dictionary – giving dictionary
parent_key (str, optional) – prefix nested keys with string parent_key
separator (str, optional) – delimiter between parent_key and key to use

Returns

ordered dictionary with flatten keys

Return type

collections.OrderedDict

catalyst.utils.misc.get_hash(obj: Any) → str[source]¶

Creates unique hash from object following way: - Represent obj as sting recursively - Hash this string with sha256 hash function - encode hash with url-safe base64 encoding

Parameters: obj – object to hash
Returns: base64-encoded string

catalyst.utils.misc.get_short_hash(obj) → str[source]¶

Creates unique short hash from object.

Parameters: obj – object to hash
Returns: short base64-encoded string (6 chars)

catalyst.utils.misc.args_are_not_none(*args: Optional[Any]) → bool[source]¶

Check that all arguments are not None.

Parameters: *args – values # noqa: RST213
Returns: True if all value were not None, False otherwise
Return type: bool

catalyst.utils.misc.make_tuple(tuple_like)[source]¶

Creates a tuple if given tuple_like value isn’t list or tuple.

Parameters: tuple_like – tuple like object - list or tuple
Returns: tuple or list

catalyst.utils.misc.pairwise(iterable: Iterable[T]) → Iterable[Tuple[T, T]][source]¶

Iterate sequences by pairs.

Examples

>>> for i in pairwise([1, 2, 5, -3]):
>>>     print(i)
(1, 2)
(2, 5)
(5, -3)

Parameters: iterable – Any iterable sequence
Returns: pairwise iterator

catalyst.utils.misc.find_value_ids(it: Iterable[Any], value: Any) → List[int][source]¶

Parameters

it – list of any
value – query element

Returns

indices of the all elements equal x0

catalyst.utils.misc.get_by_keys(dict_: dict, *keys: Any, default: Optional[T] = None) → T[source]¶: @TODO: docs.

catalyst.utils.misc.convert_labels2list(labels: Union[torch.Tensor, List[int]]) → List[int][source]¶

This function allows to work with 2 types of indexing: using a integer tensor and a list of indices.

Parameters: labels – labels of batch samples
Returns: labels of batch samples in the aligned format
Raises: TypeError – if type of input labels is not tensor and list

Numpy ¶

catalyst.utils.numpy.get_one_hot(label: int, num_classes: int, smoothing: float = None) → numpy.ndarray[source]¶

Applies OneHot vectorization to a giving scalar, optional with label smoothing as described in Bag of Tricks for Image Classification with Convolutional Neural Networks.

Parameters

label – scalar value to be vectorized
num_classes – total number of classes
smoothing (float, optional) – if specified applies label smoothing from Bag of Tricks for Image Classification with Convolutional Neural Networks paper

Returns

a one-hot vector with shape (num_classes)

Return type

np.ndarray

Onnx ¶

catalyst.utils.onnx.onnx_export(model: torch.nn.modules.module.Module, batch: torch.Tensor, file: str, method_name: str = 'forward', input_names: Iterable = None, output_names: List[str] = None, dynamic_axes: Union[Dict[str, int], Dict[str, Dict[str, int]]] = None, opset_version: int = 9, do_constant_folding: bool = False, return_model: bool = False, verbose: bool = False) → Union[None, onnx][source]¶

Converts model to onnx runtime.

Parameters

model (torch.nn.Module) – model
batch (Tensor) – inputs
file (str, optional) – file to save. Defaults to “model.onnx”.
method_name (str, optional) – Forward pass method to be converted. Defaults to “forward”.
input_names (Iterable, optional) – name of inputs in graph. Defaults to None.
output_names (List[str], optional) – name of outputs in graph. Defaults to None.
dynamic_axes (Union[Dict[str, int], Dict[str, Dict[str, int]]], optional) – axes with dynamic shapes. Defaults to None.
opset_version (int, optional) – Defaults to 9.
do_constant_folding (bool, optional) – If True, the constant-folding optimization is applied to the model during export. Defaults to False.
return_model (bool, optional) – If True then returns onnxruntime model (onnx required). Defaults to False.
verbose (bool, default False) – if specified, we will print out a debug description of the trace being exported.

Example

import torch

from catalyst.utils import convert_to_onnx

class LinModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.lin1 = torch.nn.Linear(10, 10)
        self.lin2 = torch.nn.Linear(2, 10)

    def forward(self, inp_1, inp_2):
        return self.lin1(inp_1), self.lin2(inp_2)

    def first_only(self, inp_1):
        return self.lin1(inp_1)

lin_model = LinModel()
convert_to_onnx(
    model, batch=torch.randn((1, 10)), file="model.onnx", method_name="first_only"
)

Raises: ImportError – when return_model is True, but onnx is not installed.
Returns: onnx model if return_model set to True.
Return type: Union[None, “onnx”]

catalyst.utils.onnx.quantize_onnx_model(onnx_model_path: Union[pathlib.Path, str], quantized_model_path: Union[pathlib.Path, str], qtype: str = 'qint8', verbose: bool = False) → None[source]¶

Takes model converted to onnx runtime and applies pruning.

Parameters

onnx_model_path (Union[Path, str]) – path to onnx model.
quantized_model_path (Union[Path, str]) – path to quantized model.
qtype (str, optional) – Type of weights in quantized model. Can be quint8 or qint8. Defaults to “qint8”.
verbose (bool, optional) – If set to True prints model size before and after quantization. Defaults to False.

Raises

ValueError – If qtype is not understood.

Pruning ¶

catalyst.utils.pruning.prune_model(model: torch.nn.modules.module.Module, pruning_fn: Union[Callable, str], amount: Union[float, int], keys_to_prune: Optional[List[str]] = None, layers_to_prune: Optional[List[str]] = None, dim: int = None, l_norm: int = None) → None[source]¶

Prune model function can be used for pruning certain tensors in model layers.

Parameters

model – Model to be pruned.
pruning_fn – Pruning function with API same as in torch.nn.utils.pruning. pruning_fn(module, name, amount).
keys_to_prune – list of strings. Determines which tensor in modules will be pruned.
amount – quantity of parameters to prune. If float, should be between 0.0 and 1.0 and represent the fraction of parameters to prune. If int, it represents the absolute number of parameters to prune.
layers_to_prune – list of strings - module names to be pruned. If None provided then will try to prune every module in model.
dim (int, optional) – if you are using structured pruning method you need to specify dimension. Defaults to None.
l_norm (int, optional) – if you are using ln_structured you need to specify l_norm. Defaults to None.

Example

pruned_model = prune_model(model, pruning_fn="l1_unstructured")

Raises

AttributeError – If layers_to_prune is not None, but there is no layers with specified name. OR
ValueError – if no layers have specified keys.

catalyst.utils.pruning.remove_reparametrization(model: torch.nn.modules.module.Module, keys_to_prune: List[str], layers_to_prune: Optional[List[str]] = None) → None[source]¶

Removes pre-hooks and pruning masks from the model.

Parameters

model – model to remove reparametrization.
keys_to_prune – list of strings. Determines which tensor in modules have already been pruned.
layers_to_prune – list of strings - module names have already been pruned. If None provided then will try to prune every module in model.

catalyst.utils.pruning.get_pruning_fn(pruning_fn: Union[str, Callable], dim: int = None, l_norm: int = None) → Callable[source]¶

[summary]

Parameters

pruning_fn (Union[str, Callable]) – function from torch.nn.utils.prune module or your based on BasePruningMethod. Can be string e.g. “l1_unstructured”. See pytorch docs for more details.
dim (int, optional) – if you are using structured pruning method you need to specify dimension. Defaults to None.
l_norm (int, optional) – if you are using ln_structured you need to specify l_norm. Defaults to None.

Raises

ValueError – If dim or l_norm is not defined when it’s required.

Returns

pruning_fn

Return type

Callable

Quantization ¶

catalyst.utils.quantization.quantize_model(model: torch.nn.modules.module.Module, qconfig_spec: Dict = None, dtype: Union[str, torch.dtype, None] = 'qint8') → torch.nn.modules.module.Module[source]¶

Function to quantize model weights.

Parameters

model (Model) – model to quantize
qconfig_spec (Dict, optional) – quantization config in PyTorch format. Defaults to None.
dtype (Union[str, Optional[torch.dtype]], optional) – Type of weights after quantization. Defaults to “qint8”.

Returns

quantized model

Return type

Model

Stochastic Weights Averaging (SWA)¶

catalyst.utils.swa.average_weights(state_dicts: List[dict]) → collections.OrderedDict[source]¶

Averaging of input weights.

Parameters: state_dicts – Weights to average
Raises: KeyError – If states do not match
Returns: Averaged weights

catalyst.utils.swa.get_averaged_weights_by_path_mask(path_mask: str, logdir: Union[str, pathlib.Path] = None) → collections.OrderedDict[source]¶

Averaging of input weights and saving them.

Parameters

path_mask – globe-like pattern for models to average
logdir – Path to logs directory

Returns

Averaged weights

Sys ¶

catalyst.utils.sys.dump_environment(logdir: str, config: Any = None, configs_path: List[str] = None) → None[source]¶

Saves config, environment variables and package list in JSON into logdir.

Parameters

logdir – path to logdir
config – experiment config
configs_path – path(s) to config

catalyst.utils.sys.import_module(expdir: Union[str, pathlib.Path])[source]¶

Imports python module by path.

Parameters: expdir – path to python module.
Returns: Imported module.

catalyst.utils.sys.dump_code(logdir: Union[str, pathlib.Path], expdir: Union[str, pathlib.Path] = None)[source]¶

Dumps Catalyst code for reproducibility.

Parameters

logdir – logging dir path
expdir – experiment dir path

catalyst.utils.sys.get_config_runner(expdir: pathlib.Path, config: Dict)[source]¶

Imports and creates ConfigRunner instance.

Parameters

expdir – experiment directory path
config – dictionary with experiment Config

Returns

ConfigRunner instance

Torch ¶

catalyst.utils.torch.get_optimizable_params(model_or_params)[source]¶: Returns all the parameters that requires gradients.

catalyst.utils.torch.get_optimizer_momentum(optimizer: torch.optim.optimizer.Optimizer) → float[source]¶

Get momentum of current optimizer.

Parameters: optimizer – PyTorch optimizer
Returns: momentum at first param group
Return type: float

catalyst.utils.torch.get_optimizer_momentum_list(optimizer: torch.optim.optimizer.Optimizer) → List[Optional[float]][source]¶

Get list of optimizer momentums (for each param group)

Parameters: optimizer – PyTorch optimizer
Returns: momentum for each param group
Return type: momentum_list (List[Union[float, None]])

catalyst.utils.torch.set_optimizer_momentum(optimizer: torch.optim.optimizer.Optimizer, value: float, index: int = 0)[source]¶

Set momentum of index ‘th param group of optimizer to value.

Parameters

optimizer – PyTorch optimizer
value – new value of momentum
index (int, optional) – integer index of optimizer’s param groups, default is 0

catalyst.utils.torch.get_device() → torch.device[source]¶: Simple returning the best available device (TPU > GPU > CPU).

catalyst.utils.torch.get_available_gpus()[source]¶

Array of available GPU ids.

Examples

>>> os.environ["CUDA_VISIBLE_DEVICES"] = "0,2"
>>> get_available_gpus()
[0, 2]

>>> os.environ["CUDA_VISIBLE_DEVICES"] = "0,-1,1"
>>> get_available_gpus()
[0]

>>> os.environ["CUDA_VISIBLE_DEVICES"] = ""
>>> get_available_gpus()
[]

>>> os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
>>> get_available_gpus()
[]

Returns: available GPU ids
Return type: iterable

catalyst.utils.torch.any2device(value, device: Union[str, torch.device])[source]¶

Move tensor, list of tensors, list of list of tensors, dict of tensors, tuple of tensors to target device.

Parameters

value – Object to be moved
device – target device ids

Returns

Same structure as value, but all tensors and np.arrays moved to device

catalyst.utils.torch.prepare_cudnn(deterministic: bool = None, benchmark: bool = None) → None[source]¶

Prepares CuDNN benchmark and sets CuDNN to be deterministic/non-deterministic mode

Parameters

deterministic – deterministic mode if running in CuDNN backend.
benchmark – If True use CuDNN heuristics to figure out which algorithm will be most performant for your model architecture and input. Setting it to False may slow down your training.

catalyst.utils.torch.process_model_params(model: torch.nn.modules.module.Module, layerwise_params: Dict[str, dict] = None, no_bias_weight_decay: bool = True, lr_scaling: float = 1.0) → List[Union[torch.nn.parameter.Parameter, dict]][source]¶

Gains model parameters for torch.optim.Optimizer.

Parameters

model – Model to process
layerwise_params – Order-sensitive dict where each key is regex pattern and values are layer-wise options for layers matching with a pattern
no_bias_weight_decay – If true, removes weight_decay for all bias parameters in the model
lr_scaling – layer-wise learning rate scaling, if 1.0, learning rates will not be scaled

Returns

parameters for an optimizer

Return type

iterable

Example:

>>> model = catalyst.contrib.models.segmentation.ResnetUnet()
>>> layerwise_params = collections.OrderedDict([
>>>     ("conv1.*", dict(lr=0.001, weight_decay=0.0003)),
>>>     ("conv.*", dict(lr=0.002))
>>> ])
>>> params = process_model_params(model, layerwise_params)
>>> optimizer = torch.optim.Adam(params, lr=0.0003)

catalyst.utils.torch.get_requires_grad(model: torch.nn.modules.module.Module)[source]¶

Gets the requires_grad value for all model parameters.

Example:

>>> model = SimpleModel()
>>> requires_grad = get_requires_grad(model)

Parameters: model – model
Returns: value
Return type: requires_grad

catalyst.utils.torch.set_requires_grad(model: torch.nn.modules.module.Module, requires_grad: Union[bool, Dict[str, bool]])[source]¶

Sets the requires_grad value for all model parameters.

Example:

>>> model = SimpleModel()
>>> set_requires_grad(model, requires_grad=True)
>>> # or
>>> model = SimpleModel()
>>> set_requires_grad(model, requires_grad={""})

Parameters

model – model
requires_grad – value

catalyst.utils.torch.get_network_output(net: torch.nn.modules.module.Module, *input_shapes_args, **input_shapes_kwargs)[source]¶

For each input shape returns an output tensor

Examples

>>> net = nn.Linear(10, 5)
>>> utils.get_network_output(net, (1, 10))
tensor([[[-0.2665,  0.5792,  0.9757, -0.5782,  0.1530]]])

Parameters

net – the model
*input_shapes_args – variable length argument list of shapes
**input_shapes_kwargs – key-value arguemnts of shapes

Returns

tensor with network output

catalyst.utils.torch.get_available_engine(fp16: bool = False, ddp: bool = False, amp: bool = False, apex: bool = False) → IEngine[source]¶

Returns available engine based on given arguments.

Parameters

fp16 (bool) – option to use fp16 for training. Default is False.
ddp (bool) – option to use DDP for training. Default is False.
amp (bool) – option to use APEX for training. Default is False.
apex (bool) – option to use APEX for training. Default is False.

Returns

IEngine which match requirements.

catalyst.utils.torch.detach_tensor(tensor: torch.Tensor) → numpy.ndarray[source]¶

Detach a pytorch tensor from graph and convert it to numpy array

Parameters: tensor – PyTorch tensor
Returns: numpy ndarray

catalyst.utils.torch.trim_tensors(tensors: torch.Tensor) → List[torch.Tensor][source]¶

Trim padding off of a batch of tensors to the smallest possible length. Should be used with catalyst.data.DynamicLenBatchSampler.

Adapted from Dynamic minibatch trimming to improve BERT training speed.

Parameters: tensors – list of tensors to trim.
Returns: list of trimmed tensors.
Return type: List[torch.tensor]

catalyst.utils.torch.get_optimal_inner_init(nonlinearity: torch.nn.modules.module.Module, **kwargs) → Callable[[torch.nn.modules.module.Module], None][source]¶

Create initializer for inner layers based on their activation function (nonlinearity).

Parameters

nonlinearity – non-linear activation
**kwargs – extra kwargs

Returns

optimal initialization function

Raises

NotImplementedError – if nonlinearity is out of sigmoid, tanh, relu, `leaky_relu

catalyst.utils.torch.outer_init(layer: torch.nn.modules.module.Module) → None[source]¶

Initialization for output layers of policy and value networks typically used in deep reinforcement learning literature.

Parameters: layer – torch nn.Module instance

catalyst.utils.torch.reset_weights_if_possible(module: torch.nn.modules.module.Module)[source]¶

Resets module parameters if possible.

Parameters: module – Module to reset.

catalyst.utils.torch.pack_checkpoint(model: Union[torch.nn.modules.module.Module, Dict[str, torch.nn.modules.module.Module]] = None, criterion: Union[torch.nn.modules.module.Module, Dict[str, torch.nn.modules.module.Module]] = None, optimizer: Union[torch.optim.optimizer.Optimizer, Dict[str, torch.optim.optimizer.Optimizer]] = None, scheduler: Union[torch.optim.lr_scheduler._LRScheduler, Dict[str, torch.optim.lr_scheduler._LRScheduler]] = None, **kwargs) → Dict[source]¶

Packs model, criterion, optimizer, scheduler and some extra info **kwargs to torch-based checkpoint.

Parameters

model – torch model
criterion – torch criterion
optimizer – torch optimizer
scheduler – torch scheduler
**kwargs – some extra info to pack

Returns

torch-based checkpoint with model_state_dict, criterion_state_dict, optimizer_state_dict, scheduler_state_dict keys.

catalyst.utils.torch.unpack_checkpoint(checkpoint: Dict, model: Union[torch.nn.modules.module.Module, Dict[str, torch.nn.modules.module.Module]] = None, criterion: Union[torch.nn.modules.module.Module, Dict[str, torch.nn.modules.module.Module]] = None, optimizer: Union[torch.optim.optimizer.Optimizer, Dict[str, torch.optim.optimizer.Optimizer]] = None, scheduler: Union[torch.optim.lr_scheduler._LRScheduler, Dict[str, torch.optim.lr_scheduler._LRScheduler]] = None) → None[source]¶

Load checkpoint from file and unpack the content to a model (if not None), criterion (if not None), optimizer (if not None), scheduler (if not None).

Parameters

checkpoint – checkpoint to load
model – model where should be updated state
criterion – criterion where should be updated state
optimizer – optimizer where should be updated state
scheduler – scheduler where should be updated state

catalyst.utils.torch.save_checkpoint(checkpoint: Mapping[str, Any], path: str)[source]¶

Saves checkpoint to a file.

Parameters

checkpoint – data to save.
path – filepath where checkpoint should be stored.

catalyst.utils.torch.load_checkpoint(path: str)[source]¶

Load checkpoint from path.

Parameters: path – checkpoint file to load
Returns: checkpoint content

Tracing ¶

catalyst.utils.tracing.trace_model(model: torch.nn.modules.module.Module, batch: Union[Tuple[torch.Tensor], torch.Tensor], method_name: str = 'forward') → torch.jit._script.ScriptModule[source]¶

Traces model using runner and batch.

Parameters

model – Model to trace
batch – Batch to trace the model
method_name – Model’s method name that will be used as entrypoint during tracing

Example

import torch

from catalyst.utils import trace_model

class LinModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.lin1 = torch.nn.Linear(10, 10)
        self.lin2 = torch.nn.Linear(2, 10)

    def forward(self, inp_1, inp_2):
        return self.lin1(inp_1), self.lin2(inp_2)

    def first_only(self, inp_1):
        return self.lin1(inp_1)

lin_model = LinModel()
traced_model = trace_model(
    lin_model, batch=torch.randn(1, 10), method_name="first_only"
)

Returns: Traced model
Return type: jit.ScriptModule

Contrib ¶

Compression ¶

catalyst.contrib.utils.compression.pack(data)¶

Serialize the data into bytes using pickle.

Parameters: data – a value
Returns: Returns a bytes object serialized with pickle data.

catalyst.contrib.utils.compression.pack_if_needed(data)¶

Serialize the data into bytes using pickle.

Parameters: data – a value
Returns: Returns a bytes object serialized with pickle data.

catalyst.contrib.utils.compression.unpack(bytes)¶

Deserialize bytes into an object using pickle.

Parameters: bytes – a bytes object containing serialized with pickle data.
Returns: Returns a value deserialized from the bytes-like object.

catalyst.contrib.utils.compression.unpack_if_needed(bytes)¶

Deserialize bytes into an object using pickle.

Parameters: bytes – a bytes object containing serialized with pickle data.
Returns: Returns a value deserialized from the bytes-like object.

Image ¶

catalyst.contrib.utils.image.has_image_extension(uri) → bool[source]¶

Check that file has image extension.

Parameters: uri (Union[str, pathlib.Path]) – the resource to load the file from
Returns: True if file has image extension, False otherwise
Return type: bool

catalyst.contrib.utils.image.imread(uri, grayscale: bool = False, expand_dims: bool = True, rootpath: Union[str, pathlib.Path] = None, **kwargs) → numpy.ndarray[source]¶

Reads an image from the specified file.

Parameters

uri (str, pathlib.Path, bytes, file) – the resource to load the image from, e.g. a filename, pathlib.Path, http address or file object, see imageio.imread docs for more info
grayscale – if True, make all images grayscale
expand_dims – if True, append channel axis to grayscale images rootpath (Union[str, pathlib.Path]): path to the resource with image (allows to use relative path)
rootpath (Union[str, pathlib.Path]) – path to the resource with image (allows to use relative path)
**kwargs – extra params for image read

Returns

image

Return type

np.ndarray

catalyst.contrib.utils.image.imwrite(**kwargs)[source]¶

imwrite(uri, im, format=None, **kwargs)

Write an image to the specified file. Alias for imageio.imwrite.

Parameters: **kwargs – parameters for imageio.imwrite
Returns: image save result

catalyst.contrib.utils.image.imsave(**kwargs)[source]¶

imwrite(uri, im, format=None, **kwargs)

Write an image to the specified file. Alias for imageio.imsave.

Parameters: **kwargs – parameters for imageio.imsave
Returns: image save result

catalyst.contrib.utils.image.mimread(uri, clip_range: Tuple[int, int] = None, expand_dims: bool = True, rootpath: Union[str, pathlib.Path] = None, **kwargs) → numpy.ndarray[source]¶

Reads multiple images from the specified file.

Parameters

uri (str, pathlib.Path, bytes, file) – the resource to load the image from, e.g. a filename, pathlib.Path, http address or file object, see imageio.mimread docs for more info
clip_range (Tuple[int, int]) – lower and upper interval edges, image values outside the interval are clipped to the interval edges
expand_dims – if True, append channel axis to grayscale images rootpath (Union[str, pathlib.Path]): path to the resource with image (allows to use relative path)
rootpath (Union[str, pathlib.Path]) – path to the resource with image (allows to use relative path)
**kwargs – extra params for image read

Returns

image

Return type

np.ndarray

Pandas ¶

catalyst.contrib.utils.pandas.dataframe_to_list(dataframe: pandas.core.frame.DataFrame) → List[dict][source]¶

Converts dataframe to a list of rows (without indexes).

Parameters: dataframe – input dataframe
Returns: list of rows
Return type: List[dict]

catalyst.contrib.utils.pandas.folds_to_list(folds: Union[list, str, pandas.core.series.Series]) → List[int][source]¶

This function formats string or either list of numbers into a list of unique int.

Examples

>>> folds_to_list("1,2,1,3,4,2,4,6")
[1, 2, 3, 4, 6]
>>> folds_to_list([1, 2, 3.0, 5])
[1, 2, 3, 5]

Parameters: folds (Union[list, str, pd.Series]) – Either list of numbers or one string with numbers separated by commas or pandas series
Returns: list of unique ints
Return type: List[int]
Raises: ValueError – if value in string or array cannot be casted to int

catalyst.contrib.utils.pandas.split_dataframe(dataframe: pandas.core.frame.DataFrame, train_folds: List[int], valid_folds: Optional[List[int]] = None, infer_folds: Optional[List[int]] = None, tag2class: Optional[Dict[str, int]] = None, tag_column: str = None, class_column: str = None, seed: int = 42, n_folds: int = 5) → Tuple[pandas.core.frame.DataFrame, pandas.core.frame.DataFrame, pandas.core.frame.DataFrame, pandas.core.frame.DataFrame][source]¶

Split a Pandas DataFrame into folds.

Parameters

dataframe – input dataframe
train_folds – train folds
valid_folds (List[int], optional) – valid folds. If none takes all folds not included in train_folds
infer_folds (List[int], optional) – infer folds. If none takes all folds not included in train_folds and valid_folds
tag2class (Dict[str, int], optional) – mapping from label names into int
tag_column (str, optional) – column with label names
class_column (str, optional) – column to use for split
seed – seed for split
n_folds – number of folds

Returns

tuple with 4 dataframes: whole dataframe, train part, valid part and infer part

Return type

tuple

catalyst.contrib.utils.pandas.split_dataframe_on_column_folds(dataframe: pandas.core.frame.DataFrame, column: str, random_state: int = 42, n_folds: int = 5) → pandas.core.frame.DataFrame[source]¶

Splits DataFrame into N folds.

Parameters

dataframe – a dataset
column – which column to use
random_state – seed for random shuffle
n_folds – number of result folds

Returns

new dataframe with fold column

Return type

pd.DataFrame

catalyst.contrib.utils.pandas.split_dataframe_on_folds(dataframe: pandas.core.frame.DataFrame, random_state: int = 42, n_folds: int = 5) → pandas.core.frame.DataFrame[source]¶

Splits DataFrame into N folds.

Parameters

dataframe – a dataset
random_state – seed for random shuffle
n_folds – number of result folds

Returns

new dataframe with fold column

Return type

pd.DataFrame

catalyst.contrib.utils.pandas.split_dataframe_on_stratified_folds(dataframe: pandas.core.frame.DataFrame, class_column: str, random_state: int = 42, n_folds: int = 5) → pandas.core.frame.DataFrame[source]¶

Splits DataFrame into N stratified folds.

Also see catalyst.data.sampler.BalanceClassSampler

Parameters

dataframe – a dataset
class_column – which column to use for split
random_state – seed for random shuffle
n_folds – number of result folds

Returns

new dataframe with fold column

Return type

pd.DataFrame

catalyst.contrib.utils.pandas.split_dataframe_train_test(dataframe: pandas.core.frame.DataFrame, **train_test_split_args) → Tuple[pandas.core.frame.DataFrame, pandas.core.frame.DataFrame][source]¶

Split dataframe in train and test part.

Parameters

dataframe – pd.DataFrame to split
**train_test_split_args –

test_sizefloat, int, or None (default is None)
If float, should be between 0.0 and 1.0 and represent the proportion of the dataset to include in the test split. If int, represents the absolute number of test samples. If None, the value is automatically set to the complement of the train size. If train size is also None, test size is set to 0.25.

train_sizefloat, int, or None (default is None)
If float, should be between 0.0 and 1.0 and represent the proportion of the dataset to include in the train split. If int, represents the absolute number of train samples. If None, the value is automatically set to the complement of the test size.

random_stateint or RandomState
Pseudo-random number generator state used for random sampling.

stratifyarray-like or None (default is None)
If not None, data is split in a stratified fashion, using this as the class labels.

Returns

train and test DataFrames

Note

It exist cause sklearn split is overcomplicated.

catalyst.contrib.utils.pandas.separate_tags(dataframe: pandas.core.frame.DataFrame, tag_column: str = 'tag', tag_delim: str = ', ') → pandas.core.frame.DataFrame[source]¶

Separates values in class_column column.

Parameters

dataframe – a dataset
tag_column – column name to separate values
tag_delim – delimiter to separate values

Returns

new dataframe

Return type

pd.DataFrame

catalyst.contrib.utils.pandas.map_dataframe(dataframe: pandas.core.frame.DataFrame, tag_column: str, class_column: str, tag2class: Dict[str, int], verbose: bool = False) → pandas.core.frame.DataFrame[source]¶

This function maps tags from tag_column to ints into class_column using tag2class dictionary.

Parameters

dataframe – input dataframe
tag_column – column with tags
class_column (str) –
tag2class (Dict[str, int]) – mapping from tags to class labels
verbose – flag if true, uses tqdm

Returns

updated dataframe with class_column

Return type

pd.DataFrame

catalyst.contrib.utils.pandas.get_dataset_labeling(dataframe: pandas.core.frame.DataFrame, tag_column: str) → Dict[str, int][source]¶

Prepares a mapping using unique values from tag_column.

{
    "class_name_0": 0,
    "class_name_1": 1,
    ...
    "class_name_N": N
}

Parameters

dataframe – a dataset
tag_column – which column to use

Returns

mapping from tag to labels

Return type

Dict[str, int]

catalyst.contrib.utils.pandas.balance_classes(dataframe: pandas.core.frame.DataFrame, class_column: str = 'label', random_state: int = 42, how: str = 'downsampling') → pandas.core.frame.DataFrame[source]¶

Balance classes in dataframe by class_column.

Parameters

dataframe – a dataset
class_column – which column to use for split
random_state – seed for random shuffle
how – strategy to sample, must be one on [“downsampling”, “upsampling”]

Returns

new dataframe with balanced class_column

Return type

pd.DataFrame

Raises

NotImplementedError – if how is not in [“upsampling”, “downsampling”, int]

catalyst.contrib.utils.pandas.create_dataset(dirs: str, extension: str = None, process_fn: Callable[[str], object] = None, recursive: bool = False) → Dict[str, List[Any]][source]¶

Create dataset (dict like {key: [values]}) from vctk-like dataset:

dataset/
    cat/
        *.ext
    dog/
        *.ext

Parameters

dirs – path to dirs, for example /home/user/data/**
extension – data extension you are looking for
process_fn (Callable[[str], object]) – function(path_to_file) -> object process function for found files, by default
recursive – enables recursive globbing

Returns

dataset

Return type

dict

catalyst.contrib.utils.pandas.create_dataframe(dataset: Dict[str, List[Any]], **dataframe_args) → pandas.core.frame.DataFrame[source]¶

Create pd.DataFrame from dict like {key: [values]}.

Parameters

dataset – dict like {key: [values]}
**dataframe_args –

indexIndex or array-like
Index to use for resulting frame. Will default to np.arange(n) if no indexing information part of input data and no index provided

columnsIndex or array-like
Column labels to use for resulting frame. Will default to np.arange(n) if no column labels are provided

dtypedtype, default None
Data type to force, otherwise infer

Returns

dataframe from giving dataset

Return type

pd.DataFrame

catalyst.contrib.utils.pandas.split_dataset_train_test(dataset: pandas.core.frame.DataFrame, **train_test_split_args) → Tuple[Dict[str, List[Any]], Dict[str, List[Any]]][source]¶

Split dataset in train and test parts.

Parameters

dataset – dict like dataset
**train_test_split_args –

test_sizefloat, int, or None (default is None)
If float, should be between 0.0 and 1.0 and represent the proportion of the dataset to include in the test split. If int, represents the absolute number of test samples. If None, the value is automatically set to the complement of the train size. If train size is also None, test size is set to 0.25.

train_sizefloat, int, or None (default is None)
If float, should be between 0.0 and 1.0 and represent the proportion of the dataset to include in the train split. If int, represents the absolute number of train samples. If None, the value is automatically set to the complement of the test size.

random_stateint or RandomState
Pseudo-random number generator state used for random sampling.

stratifyarray-like or None (default is None)
If not None, data is split in a stratified fashion, using this as the class labels.

Returns

train and test dicts

Parallel ¶

catalyst.contrib.utils.parallel.parallel_imap(func, args, pool: Union[multiprocessing.pool.Pool, catalyst.contrib.utils.parallel.DumbPool]) → List[T][source]¶: @TODO: Docs. Contribution is welcome.

catalyst.contrib.utils.parallel.tqdm_parallel_imap(func, args, pool: Union[multiprocessing.pool.Pool, catalyst.contrib.utils.parallel.DumbPool], total: int = None, pbar=<class 'tqdm.asyncio.tqdm_asyncio'>) → List[T][source]¶: @TODO: Docs. Contribution is welcome.

catalyst.contrib.utils.parallel.get_pool(workers: int) → Union[multiprocessing.pool.Pool, catalyst.contrib.utils.parallel.DumbPool][source]¶: @TODO: Docs. Contribution is welcome.

Serialization ¶

catalyst.contrib.utils.serialization.serialize(data)¶

Serialize the data into bytes using pickle.

Parameters: data – a value
Returns: Returns a bytes object serialized with pickle data.

catalyst.contrib.utils.serialization.deserialize(bytes)¶

Deserialize bytes into an object using pickle.

Parameters: bytes – a bytes object containing serialized with pickle data.
Returns: Returns a value deserialized from the bytes-like object.

Visualization ¶

catalyst.contrib.utils.visualization.plot_confusion_matrix(cm: numpy.ndarray, class_names=None, normalize=False, title='confusion matrix', fname=None, show=True, figsize=12, fontsize=32, colormap='Blues')[source]¶

Render the confusion matrix and return matplotlib”s figure with it. Normalization can be applied by setting normalize=True.

Parameters

cm – numpy confusion matrix
class_names – class names
normalize – boolean flag to normalize confusion matrix
title – title
fname – filename to save confusion matrix
show – boolean flag for preview
figsize – matplotlib figure size
fontsize – matplotlib font size
colormap – matplotlib color map

Returns

matplotlib figure

catalyst.contrib.utils.visualization.render_figure_to_array(figure)[source]¶: Renders matplotlib”s figure to tensor.