utils/buffer.py

# Copyright 2022-present, Lorenzo Bonicelli, Pietro Buzzega, Matteo Boschini, Angelo Porrello, Simone Calderara.
# All rights reserved.
# This source code is licensed under the license found in the
# LICENSE file in the root directory of this source tree.

import torch
import torch.nn as nn
import numpy as np
from typing import Tuple
from torchvision import transforms
from copy import deepcopy

def icarl_replay(self, train_loader, val_set_split=0):
    """
    Merge the replay buffer with the current task data.
    Optionally split the replay buffer into a validation set.

    :param self: the model instance
    :param dataset: the dataset
    :param val_set_split: the fraction of the replay buffer to be used as validation set
    """
        
    if self.task_id > 0:
        buff_val_mask = torch.rand(len(self.buffer)) < val_set_split
        val_train_mask = torch.zeros(len(train_loader.dataset.data)).bool()
        val_train_mask[torch.randperm(len(train_loader.dataset.data))[:buff_val_mask.sum()]] = True

        if val_set_split > 0:
            self.val_loader = deepcopy(train_loader)
        
        data_concatenate = torch.cat if type(train_loader.dataset.data) == torch.Tensor else np.concatenate
        need_aug = hasattr(train_loader.dataset, 'not_aug_transform')
        if not need_aug:
            refold_transform = lambda x: x.cpu()
        else:    
            data_shape = len(train_loader.dataset.data[0].shape)
            if data_shape == 3:
                refold_transform = lambda x: (x.cpu()*255).permute([0, 2, 3, 1]).numpy().astype(np.uint8)
            elif data_shape == 2:
                refold_transform = lambda x: (x.cpu()*255).squeeze(1).type(torch.uint8)

        # REDUCE AND MERGE TRAINING SET
        train_loader.dataset.targets = np.concatenate([
            train_loader.dataset.targets[~val_train_mask],
            self.buffer.labels.cpu().numpy()[:len(self.buffer)][~buff_val_mask]
            ])
        train_loader.dataset.data = data_concatenate([
            train_loader.dataset.data[~val_train_mask],
            refold_transform((self.buffer.examples)[:len(self.buffer)][~buff_val_mask])
            ])

        if val_set_split > 0:
            # REDUCE AND MERGE VALIDATION SET
            self.val_loader.dataset.targets = np.concatenate([
                self.val_loader.dataset.targets[val_train_mask],
                self.buffer.labels.cpu().numpy()[:len(self.buffer)][buff_val_mask]
                ])
            self.val_loader.dataset.data = data_concatenate([
                self.val_loader.dataset.data[val_train_mask],
                refold_transform((self.buffer.examples)[:len(self.buffer)][buff_val_mask])
                ])

class Buffer(nn.Module):
    """
    The memory buffer of rehearsal method.
    """
    def __init__(self, buffer_size, device, num_seen_examples=0, blocked=False, n_tasks=None, mode='reservoir', class_balance=False):
        super().__init__()
        assert mode in ['ring', 'reservoir']
        self.buffer_size = buffer_size
        self.device = device
        
        self.class_balance = class_balance
        if self.class_balance: print("Using class balanced buffer")
        else: print("Using standard reservoir buffer")

        self.num_seen_examples = num_seen_examples
        if mode == 'ring':
            assert n_tasks is not None
            self.task_number = n_tasks
            self.buffer_portion_size = buffer_size // n_tasks
        self.attributes = ['examples', 'labels', 'logits', 'task_labels', 'grad_inputs', "final_feats","logits2","final_feats2","mask"]

        self.blocked = blocked

    def get_extra_state(self):
        state = {
            'num_seen_examples': self.num_seen_examples,
            'blocked': self.blocked
        }
        return state

    def set_extra_state(self, state):
        self.num_seen_examples = state['num_seen_examples']
        self.blocked = state['blocked']

    def turn_on_blocking(self):
        self.blocked = True
    
    def turn_off_blocking(self):
        self.blocked = False

    def reservoir(self, num_seen_examples: int, buffer_size: int) -> int:
        """
        Reservoir sampling algorithm.
        :param num_seen_examples: the number of seen examples
        :param buffer_size: the maximum buffer size
        :return: the target index if the current image is sampled, else -1
        """
        if num_seen_examples < buffer_size:
            return num_seen_examples

        rand = np.random.randint(0, num_seen_examples + 1)
        if rand < buffer_size:
            return rand
        else:
            return -1

    def reservoir_balanced(self, num_seen_examples:int, buffer_size:int) -> int:
        if num_seen_examples < buffer_size:
            return num_seen_examples

        rand = np.random.randint(0, num_seen_examples + 1)
        if rand < buffer_size:
            # Return a random index corresponding to classes that have most examples in the buffer
            y = self.labels
            classes, counts = torch.unique(y, return_counts = True)
            max_count = counts.max()
            classes_max = classes[counts == max_count]
            idx_max = torch.stack([y==c for c in classes_max], dim=0).float().sum(0)
            idx_max = (idx_max > 0).nonzero().squeeze()
            idx = np.random.choice(idx_max.cpu().numpy())
            
            return idx
        else:
            return -1


    def ring(self, num_seen_examples: int, buffer_portion_size: int, task: int) -> int:
        return num_seen_examples % buffer_portion_size + task * buffer_portion_size

    def to(self, device):
        self.device = device
        for attr_str in self.attributes:
            if hasattr(self, attr_str):
                setattr(self, attr_str, getattr(self, attr_str).to(device))
        return self

    def __len__(self):
        return min(self.num_seen_examples, self.buffer_size)


    def init_tensors(self, examples: torch.Tensor, labels: torch.Tensor,
                     logits: torch.Tensor, task_labels: torch.Tensor, 
                     grad_inputs:torch.Tensor, final_feats:torch.Tensor,logits2: torch.Tensor,final_feats2:torch.Tensor,mask:torch.Tensor,
                     ) -> None:
        """
        Initializes just the required tensors.
        :param examples: tensor containing the images
        :param labels: tensor containing the labels
        :param logits: tensor containing the outputs of the network
        :param task_labels: tensor containing the task labels
        """
        for attr_str in self.attributes:
            attr = eval(attr_str)
            if attr is not None and not hasattr(self, attr_str):
                typ = torch.int64 if attr_str.endswith('els') else torch.float32
                tensor = torch.zeros((self.buffer_size,
                    *attr.shape[1:]), dtype=typ, device=self.device)
                # setattr(self, attr_str, tensor)
                self.register_buffer(attr_str, tensor)

    def get_index_reservoir(self, examples):
        '''
        Get the insertion index of the given examples, using reservoir sampling
        :param examples: the examples to insert
        :return: the insertion index
        '''
        index = []
        for i in range(len(examples)):
            if self.class_balance:
                idx = self.reservoir_balanced(self.num_seen_examples, self.buffer_size)
            else:
                idx = self.reservoir(self.num_seen_examples, self.buffer_size)
            index.append(idx)
            self.num_seen_examples += 1
        return index

    def add_data(self, examples, indices=None, 
                 labels=None, logits=None, task_labels=None, grad_inputs=None, final_feats=None,logits2=None,final_feats2=None,mask=None):
        """
        Adds the data to the memory buffer according to the reservoir strategy.
        :param examples: tensor containing the images
        :param indices: indices at which the examples should be inserted
        :param labels: tensor containing the labels
        :param logits: tensor containing the outputs of the network
        :param task_labels: tensor containing the task labels
        :param grad_inputs: tensor containing the gradients of output w.r.t. of the inputs
        :return:
        """
        if self.blocked:
            return
        
        if not hasattr(self, 'examples'):
            self.init_tensors(examples, labels, logits, task_labels, grad_inputs, final_feats, logits2,final_feats2,mask)

        for i in range(examples.shape[0]):
            if indices is not None:
                index = indices[i]
            elif self.class_balance:
                index = self.reservoir_balanced(self.num_seen_examples, self.buffer_size)
            else:
                index = self.reservoir(self.num_seen_examples, self.buffer_size)

            self.num_seen_examples += 1
            if index >= 0:
                self.examples[index] = examples[i].to(self.device)
                if labels is not None:
                    self.labels[index] = labels[i].to(self.device)
                if logits is not None:
                    self.logits[index] = logits[i].to(self.device)
                if task_labels is not None:
                    self.task_labels[index] = task_labels[i].to(self.device)
                if grad_inputs is not None:
                    self.grad_inputs[index] = grad_inputs[i].to(self.device)
                if final_feats is not None:
                    self.final_feats[index] = final_feats[i].to(self.device)
                if logits2 is not None:
                    self.logits2[index] = logits2[i].to(self.device)
                if final_feats2 is not None:
                    self.final_feats2[index] = final_feats2[i].to(self.device)
                if mask is not None:
                    self.mask[index] = mask[i].to(self.device)


    def get_data(self, size: int, transform: transforms=None, return_index=False) -> Tuple:
        """
        Random samples a batch of size items.
        :param size: the number of requested items
        :param transform: the transformation to be applied (data augmentation)
        :return:
        """
        if size > min(self.num_seen_examples, self.examples.shape[0]):
            size = min(self.num_seen_examples, self.examples.shape[0])

        choice = np.random.choice(min(self.num_seen_examples, self.examples.shape[0]),
                                  size=size, replace=False)
        if transform is None: transform = lambda x: x
        ret_tuple = (torch.stack([transform(ee.cpu())
                            for ee in self.examples[choice]]).to(self.device),)
        for attr_str in self.attributes[1:]:
            if hasattr(self, attr_str):
                attr = getattr(self, attr_str)
                ret_tuple += (attr[choice],)
        
        if not return_index:
            return ret_tuple
        else:
            return (torch.tensor(choice).to(self.device), ) + ret_tuple

    def get_data_by_index(self, indexes, transform: transforms=None) -> Tuple:
        """
        Returns the data by the given index.
        :param index: the index of the item
        :param transform: the transformation to be applied (data augmentation)
        :return:
        """
        if transform is None: transform = lambda x: x
        ret_tuple = (torch.stack([transform(ee.cpu())
                            for ee in self.examples[indexes]]).to(self.device),)
        for attr_str in self.attributes[1:]:
            if hasattr(self, attr_str):
                attr = getattr(self, attr_str).to(self.device)
                ret_tuple += (attr[indexes],)
        return ret_tuple


    def is_empty(self) -> bool:
        """
        Returns true if the buffer is empty, false otherwise.
        """
        if self.num_seen_examples == 0 or not hasattr(self, "examples"):
            return True
        else:
            return False

    def get_all_data(self, transform: transforms=None) -> Tuple:
        """
        Return all the items in the memory buffer.
        :param transform: the transformation to be applied (data augmentation)
        :return: a tuple with all the items in the memory buffer
        """
        if transform is None: transform = lambda x: x
        ret_tuple = (torch.stack([transform(ee.cpu())
                            for ee in self.examples]).to(self.device),)
        for attr_str in self.attributes[1:]:
            if hasattr(self, attr_str):
                attr = getattr(self, attr_str)
                ret_tuple += (attr,)
        return ret_tuple

    def empty(self) -> None:
        """
        Set all the tensors to None.
        """
        for attr_str in self.attributes:
            if hasattr(self, attr_str):
                delattr(self, attr_str)
        self.num_seen_examples = 0