Utils

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]

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

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

Distributed

catalyst.utils.distributed.get_backend() → Optional[str]

Returns the backend for distributed training.

catalyst.utils.distributed.get_rank() → int

Returns the rank of the current worker.

Returns

rank if torch.distributed is initialized, otherwise -1

Return type

int

catalyst.utils.distributed.get_world_size() → int

Returns the world size for distributed training.

catalyst.utils.distributed.get_distributed_params()

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

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

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

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]

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.

catalyst.utils.distributed.ddp_reduce(tensor: torch.Tensor, mode: str, world_size: int)

Syncs tensor over world_size in distributed mode.

Parameters

• tensor – tensor to sync across the processes.

• mode – tensor synchronization type, should be one of ‘sum’, ‘mean’ or ‘all’.

• world_size – world size

Returns

torch.Tensor with synchronized values.

Raises

ValueError – if mode is out of sum, mean, all.

catalyst.utils.distributed.ddp_sync_run(function: Callable)

Runs function in a synchronous way: 0-rank first and all other processes after.

Parameters

function – callable function

Misc

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

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)

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)

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

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

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)

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

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

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

Parameters

seed – random seed

catalyst.utils.misc.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.

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

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

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

Creates unique short hash from object.

Parameters

obj – object to hash

Returns

short base64-encoded string (6 chars)

catalyst.utils.misc.make_tuple(tuple_like)

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]]

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.get_by_keys(dict_: dict, *keys: Any, default: Optional[T] = None) → T

@TODO: docs.

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]

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

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

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

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

[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

Function to quantize model weights.

Parameters

• model – model to be quantized

• 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

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

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

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])

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)

Dumps Catalyst code for reproducibility.

Parameters

• logdir – logging dir path

• expdir – experiment dir path

Torch

catalyst.utils.torch.get_optimizable_params(model_or_params)

Returns all the parameters that requires gradients.

catalyst.utils.torch.get_optimizer_momentum(optimizer: torch.optim.optimizer.Optimizer) → float

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]]

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)

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

Simple returning the best available device (TPU > GPU > CPU).

catalyst.utils.torch.get_available_gpus()

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: Union[Dict, List, Tuple, numpy.ndarray, torch.Tensor, torch.nn.modules.module.Module], device: Union[str, torch.device]) → Union[Dict, List, Tuple, torch.Tensor, torch.nn.modules.module.Module]

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

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]]

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)

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]])

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)

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

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

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]

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]

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

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)

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

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

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)

Saves checkpoint to a file.

Parameters

• checkpoint – data to save.

• path – filepath where checkpoint should be stored.

catalyst.utils.torch.load_checkpoint(path: str)

Load checkpoint from path.

Parameters

path – checkpoint file to load

Returns

checkpoint content

catalyst.utils.torch.soft_update(target: torch.nn.modules.module.Module, source: torch.nn.modules.module.Module, tau: float) → None

Updates the target data with the source one smoothing by tau (inplace operation).

Parameters

• target – nn.Module to update

• source – nn.Module for updating

• tau – smoothing parametr

catalyst.utils.torch.mixup_batch(batch: List[torch.Tensor], alpha: float = 0.2, mode: str = 'replace') → List[torch.Tensor]

Parameters

• batch – batch to which you want to apply augmentation

• alpha – beta distribution a=b parameters. Must be >=0. The closer alpha to zero the less effect of the mixup.

• mode – algorithm used for muxup: "replace" | "add". If “replace” then replaces the batch with a mixed one, while the batch size is not changed If “add”, concatenates mixed examples to the current ones, the batch size increases by 2 times.

Returns

augmented batch

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

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

Image

catalyst.contrib.utils.image.has_image_extension(uri) → bool

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

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)

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)

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

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

Report

catalyst.contrib.utils.report.get_classification_report(y_true: numpy.ndarray, y_pred: numpy.ndarray, y_scores: numpy.ndarray = None, beta: float = None) → pandas.core.frame.DataFrame

Generates pandas-based per-class and aggregated classification metrics.

Parameters

• y_true (np.ndarray) – ground truth labels

• y_pred (np.ndarray) – predicted model labels

• y_scores (np.ndarray) – predicted model scores. Defaults to None.

• beta (float, optional) – Beta parameter for custom Fbeta score computation. Defaults to None.

Returns

pandas dataframe with main classification metrics.

Return type

pd.DataFrame

Examples:

from sklearn import datasets, linear_model, metrics
from sklearn.model_selection import train_test_split
from catalyst import utils

digits = datasets.load_digits()

# flatten the images
n_samples = len(digits.images)
data = digits.images.reshape((n_samples, -1))

# Create a classifier
clf = linear_model.LogisticRegression(multi_class="ovr")

# Split data into 50% train and 50% test subsets
X_train, X_test, y_train, y_test = train_test_split(
    data, digits.target, test_size=0.5, shuffle=False)

# Learn the digits on the train subset
clf.fit(X_train, y_train)

# Predict the value of the digit on the test subset
y_scores = clf.predict_proba(X_test)
y_pred = clf.predict(X_test)

utils.get_classification_report(
    y_true=y_test,
    y_pred=y_pred,
    y_scores=y_scores,
    beta=0.5
)

Thresholds

catalyst.contrib.utils.thresholds.get_baseline_thresholds(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float]) → Tuple[float, List[float]]

Returns baseline thresholds for multiclass/multilabel classification.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model, numpy array with shape [num_examples, num_classes]

• labels – ground truth labels, numpy array with shape [num_examples, num_classes]

• objective – callable function, metric which we want to maximize

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_binary_threshold(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float], num_thresholds: int = 100) → Tuple[float, float]

Finds best threshold for binary classification task based on cross-validation estimates.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model, numpy array with shape [num_examples, ]

• labels – ground truth labels, numpy array with shape [num_examples, ]

• objective – callable function, metric which we want to maximize

• num_thresholds – number of thresholds ot try for each class

Returns

tuple with best found objective score and threshold

catalyst.contrib.utils.thresholds.get_multiclass_thresholds(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float]) → Tuple[List[float], List[float]]

Finds best thresholds for multiclass classification task.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model, numpy array with shape [num_examples, num_classes]

• labels – ground truth labels, numpy array with shape [num_examples, num_classes]

• objective – callable function, metric which we want to maximize

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_multilabel_thresholds(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float])

Finds best thresholds for multilabel classification task.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model, numpy array with shape [num_examples, num_classes]

• labels – ground truth labels, numpy array with shape [num_examples, num_classes]

• objective – callable function, metric which we want to maximize

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_binary_threshold_cv(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float], num_splits: int = 5, num_repeats: int = 1, random_state: int = 42)

Finds best threshold for binary classification task based on cross-validation estimates.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model, numpy array with shape [num_examples, ]

• labels – ground truth labels, numpy array with shape [num_examples, ]

• objective – callable function, metric which we want to maximize

• num_splits – number of splits to use for cross-validation

• num_repeats – number of repeats to use for cross-validation

• random_state – random state to use for cross-validation

Returns

tuple with best found objective score and threshold

catalyst.contrib.utils.thresholds.get_multilabel_thresholds_cv(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float], num_splits: int = 5, num_repeats: int = 1, random_state: int = 42)

Finds best thresholds for multilabel classification task based on cross-validation estimates.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model, numpy array with shape [num_examples, num_classes]

• labels – ground truth labels, numpy array with shape [num_examples, num_classes]

• objective – callable function, metric which we want to maximize

• num_splits – number of splits to use for cross-validation

• num_repeats – number of repeats to use for cross-validation

• random_state – random state to use for cross-validation

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_thresholds_greedy(scores: numpy.ndarray, labels: numpy.ndarray, score_fn: Callable, num_iterations: int = 100, num_thresholds: int = 100, thresholds: numpy.ndarray = None, patience: int = 3, atol: float = 0.01) → Tuple[float, List[float]]

Finds best thresholds for classification task with brute-force algorithm.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model

• labels – ground truth labels

• score_fn – callable function, based on (scores, labels, thresholds)

• num_iterations – number of iteration for brute-force algorithm

• num_thresholds – number of thresholds ot try for each class

• thresholds – baseline thresholds, which we want to optimize

• patience – maximum number of iteration before early stop exit

• atol – minimum required improvement per iteration for early stop exit

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_multilabel_thresholds_greedy(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float], num_iterations: int = 100, num_thresholds: int = 100, thresholds: numpy.ndarray = None, patience: int = 3, atol: float = 0.01) → Tuple[float, List[float]]

Finds best thresholds for multilabel classification task with brute-force algorithm.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model

• labels – ground truth labels

• objective – callable function, metric which we want to maximize

• num_iterations – number of iteration for brute-force algorithm

• num_thresholds – number of thresholds ot try for each class

• thresholds – baseline thresholds, which we want to optimize

• patience – maximum number of iteration before early stop exit

• atol – minimum required improvement per iteration for early stop exit

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_multiclass_thresholds_greedy(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float], num_iterations: int = 100, num_thresholds: int = 100, thresholds: numpy.ndarray = None, patience: int = 3, atol: float = 0.01) → Tuple[float, List[float]]

Finds best thresholds for multiclass classification task with brute-force algorithm.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model

• labels – ground truth labels

• objective – callable function, metric which we want to maximize

• num_iterations – number of iteration for brute-force algorithm

• num_thresholds – number of thresholds ot try for each class

• thresholds – baseline thresholds, which we want to optimize

• patience – maximum number of iteration before early stop exit

• atol – minimum required improvement per iteration for early stop exit

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_best_multilabel_thresholds(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float]) → Tuple[float, List[float]]

Finds best thresholds for multilabel classification task.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model

• labels – ground truth labels

• objective – callable function, metric which we want to maximize

Returns

tuple with best found objective score and per-class thresholds

catalyst.contrib.utils.thresholds.get_best_multiclass_thresholds(scores: numpy.ndarray, labels: numpy.ndarray, objective: Callable[[numpy.ndarray, numpy.ndarray], float]) → Tuple[float, List[float]]

Finds best thresholds for multiclass classification task.

Parameters

• scores – estimated per-class scores/probabilities predicted by the model

• labels – ground truth labels

• objective – callable function, metric which we want to maximize

Returns

tuple with best found objective score and per-class thresholds

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')

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)

Renders matplotlib”s figure to tensor.