Utils

utils.binary_to_boolean_type(value)

Converts a binary string to a boolean type.

Parameters:

value (str) – the binary string

Returns:

the boolean type

Return type:

bool

utils.create_if_not_exists(path)

Creates the specified folder if it does not exist.

Parameters:

path (str) – the complete path of the folder to be created

utils.infer_args_from_signature(signature, excluded_signature=None)

Load the arguments of a function from its signature.

Parameters:

signature (Signature) – the signature of the function

Returns:

the inferred arguments

Return type:

dict

utils.register_dynamic_module_fn(name, register)

Register a dynamic module in the specified dictionary.

Parameters:

• name (str) – the name of the module

• register (dict) – the dictionary where the module will be registered

• cls – the class to be registered

• tp – the type of the class, used to dynamically infer the arguments

Submodules

Args

Buffer

class utils.buffer.BaseSampleSelection(buffer_size, device)

Bases: object

Base class for sample selection strategies.

update(*args, **kwargs)

(optional) Update the state of the sample selection strategy.

class utils.buffer.Buffer(buffer_size, device='cpu')

Bases: object

The memory buffer of rehearsal method.

add_data(examples, labels=None, logits=None, task_labels=None, true_labels=None, sample_selection_scores=None)

Adds the data to the memory buffer according to the reservoir strategy.

Parameters:

• examples – tensor containing the images

• labels – tensor containing the labels

• logits – tensor containing the outputs of the network

• task_labels – tensor containing the task labels

• true_labels – if setting is noisy, the true labels associated with the examples. Used only for logging.

• sample_selection_scores – tensor containing the scores used for the sample selection strategy. NOTE: this is only used if the sample selection strategy defines the update method.

Note

Only the examples are required. The other tensors are initialized only if they are provided.

attributes: List[str]

buffer_size: int

device: str | device

examples: Tensor

get_data(size, transform=None, device=None)

Random samples a batch of size items.

Parameters:

• size (int) – the number of requested items

• transform (Module | None) – the transformation to be applied (data augmentation)

• device (str | device | None) – the device to store the data on. If None, uses the device of the buffer.

Returns:

a tuple containing the requested items. If return_index is True, the tuple contains the indexes as first element.

Return type:

Sequence[Tensor]

init_tensors(examples, labels, logits, task_labels, true_labels)

Initializes just the required tensors.

Parameters:

• examples (Tensor) – tensor containing the images

• labels (Tensor) – tensor containing the labels

• logits (Tensor) – tensor containing the outputs of the network

• task_labels (Tensor) – tensor containing the task labels

• true_labels (Tensor) – tensor containing the true labels (used only for logging)

is_empty()

Returns true if the buffer is empty, false otherwise.

Return type:

bool

is_full()

labels: Tensor

logits: Tensor

num_seen_examples: int

reset()

Set all the tensors to None.

serialize(out_device='cpu')

Serialize the buffer.

Returns:

A dictionary containing the buffer attributes.

task_labels: Tensor

to(device)

Move the buffer and its attributes to the specified device.

Parameters:

device – The device to move the buffer and its attributes to.

Returns:

The buffer instance with the updated device and attributes.

true_labels: Tensor

property used_attributes

Returns a list of attributes that are currently being used by the object.

class utils.buffer.ReservoirSampling(buffer_size, device)

Bases: BaseSampleSelection

Checkpoints

utils.checkpoints.mammoth_load_checkpoint(args, model)

Loads the keys from the given checkpoint. - Handles DataParallel and DistributedDataParallel checkpoints. - Handles checkpoints from previous versions of the code. - Handles head initialization for LUCIR.

Parameters:

• args – the model with the checkpoint loaded.

• model (ContinualModel) – the model to be loaded.

• ignore_classifier – whether to ignore the classifier weights.

Returns:

the model with the checkpoint loaded.

Return type:

Tuple[ContinualModel, Tuple[List[float], List[float]]]

utils.checkpoints.save_mammoth_checkpoint(task, end_task, args, model, results, optimizer_st)

Save a checkpoint for the model for the given task. Handles saving as a single file (will require weights_only=False) or separate weights (can be loaded safely with weights_only=True).

utils.checkpoints.to_parsable_obj(r)

Convert a non-builtin object to a parsable (and loadable with weights_only=True) object. Looking at you, Namespace.

Return type:

Dict | list | str | int | float | bool

Configuration

This module contains utility functions for configuration settings.

utils.conf.base_path()

Returns the base path where to save data.

Returns:

./data/)

Return type:

the base path (default

utils.conf.get_device()

Returns the least used GPU device if available else MPS or CPU.

Return type:

device

Evaluation

utils.evaluate.evaluate(model, dataset, last_only=False)

Evaluates the single-class top-1 accuracy of the model for each past task.

The accuracy is evaluated for all the tasks up to the current one, only for the total number of classes seen so far.

Parameters:

• model (ContinualModel) – the model to be evaluated

• dataset (ContinualDataset) – the continual dataset at hand

• last_only – if True, only the last task is evaluated. Defaults to False.

Returns:

a tuple of lists, containing the class-il and task-il accuracy for each task.

Return type:

Tuple[list, list]

utils.evaluate.mask_classes(outputs, dataset, k)

Given the output tensor, the dataset at hand and the current task, masks the former by setting the responses for the other tasks at -inf. It is used to obtain the results for the task-il setting.

Parameters:

• outputs (Tensor) – the output tensor

• dataset (ContinualDataset) – the continual dataset

• k (int) – the task index

Training

utils.training.train(model, dataset, args=None)

The training process, including evaluations and loggers.

Parameters:

• model (ContinualModel) – the module to be trained

• dataset (ContinualDataset) – the continual dataset at hand

• args (Namespace | None) – the arguments of the current execution. If None, it will be loaded from the environment variable ‘MAMMOTH_ARGS’.

utils.training.train_epoch(model, train_loader, args, epoch, pbar)

Trains the model for a single epoch.

Parameters:

• model (ContinualModel) – the model to be trained

• train_loader (Iterable) – the data loader for the training set

• args (Namespace) – the arguments from the command line

• epoch (int) – the current epoch

• pbar (tqdm) – the progress bar to update

Returns:

the number of iterations performed in the current epoch

utils.training.update_progress_bar(pbar, model)

Utils