| | |
| | """ |
| | @inproceedings{DBLP:conf/cvpr/HouPLWL19, |
| | title = {Learning a Unified Classifier Incrementally via Rebalancing}, |
| | author = {Saihui Hou and |
| | Xinyu Pan and |
| | Chen Change Loy and |
| | Zilei Wang and |
| | Dahua Lin}, |
| | booktitle = {{IEEE} Conference on Computer Vision and Pattern Recognition, {CVPR} |
| | 2019, Long Beach, CA, USA, June 16-20, 2019}, |
| | pages = {831--839}, |
| | publisher = {Computer Vision Foundation / {IEEE}}, |
| | year = {2019} |
| | } |
| | https://openaccess.thecvf.com/content_CVPR_2019/html/Hou_Learning_a_Unified_Classifier_Incrementally_via_Rebalancing_CVPR_2019_paper.html |
| | |
| | Adapted from https://github.com/hshustc/CVPR19_Incremental_Learning |
| | """ |
| |
|
| | import math |
| | import copy |
| | import torch |
| | import torch.nn as nn |
| | from torch.nn import Parameter |
| | import torch.nn.functional as F |
| | from .finetune import Finetune |
| | from core.model.backbone.resnet import * |
| | import numpy as np |
| | from torch.utils.data import DataLoader |
| |
|
| |
|
| | cur_features = [] |
| | ref_features = [] |
| | old_scores = [] |
| | new_scores = [] |
| | def get_ref_features(self, inputs, outputs): |
| | global ref_features |
| | ref_features = inputs[0] |
| |
|
| | def get_cur_features(self, inputs, outputs): |
| | global cur_features |
| | cur_features = inputs[0] |
| |
|
| | def get_old_scores_before_scale(self, inputs, outputs): |
| | global old_scores |
| | old_scores = outputs |
| |
|
| | def get_new_scores_before_scale(self, inputs, outputs): |
| | global new_scores |
| | new_scores = outputs |
| |
|
| |
|
| |
|
| | class Model(nn.Module): |
| | |
| | def __init__(self, backbone, feat_dim, num_class): |
| | super().__init__() |
| | self.backbone = backbone |
| | self.feat_dim = feat_dim |
| | self.num_class = num_class |
| | self.classifier = CosineLinear(feat_dim, num_class) |
| | |
| | def forward(self, x): |
| | return self.get_logits(x) |
| | |
| | def get_logits(self, x): |
| | logits = self.classifier(self.backbone(x)['features']) |
| | return logits |
| |
|
| |
|
| |
|
| | class LUCIR(Finetune): |
| | def __init__(self, backbone, feat_dim, num_class, **kwargs): |
| | super().__init__(backbone, feat_dim, num_class, **kwargs) |
| | self.kwargs = kwargs |
| | self.network = Model(self.backbone, feat_dim, kwargs['init_cls_num']) |
| | self.K = kwargs['K'] |
| | self.lw_mr = kwargs['lw_mr'] |
| | self.ref_model = None |
| | self.task_idx = 0 |
| |
|
| | def before_task(self, task_idx, buffer, train_loader, test_loaders): |
| | self.task_idx = task_idx |
| |
|
| | if task_idx == 1: |
| | self.ref_model = copy.deepcopy(self.network) |
| | in_features = self.network.classifier.in_features |
| | out_features = self.network.classifier.out_features |
| | new_fc = SplitCosineLinear(in_features, out_features, self.kwargs['inc_cls_num']) |
| | new_fc.fc1.weight.data = self.network.classifier.weight.data |
| | new_fc.sigma.data = self.network.classifier.sigma.data |
| | self.network.classifier = new_fc |
| | lamda_mult = out_features*1.0 / self.kwargs['inc_cls_num'] |
| |
|
| |
|
| | elif task_idx > 1: |
| | self.ref_model = copy.deepcopy(self.network) |
| | in_features = self.network.classifier.in_features |
| | out_features1 = self.network.classifier.fc1.out_features |
| | out_features2 = self.network.classifier.fc2.out_features |
| | new_fc = SplitCosineLinear(in_features, out_features1+out_features2, self.kwargs['inc_cls_num']).to(self.device) |
| | new_fc.fc1.weight.data[:out_features1] = self.network.classifier.fc1.weight.data |
| | new_fc.fc1.weight.data[out_features1:] = self.network.classifier.fc2.weight.data |
| | new_fc.sigma.data = self.network.classifier.sigma.data |
| | self.network.classifier = new_fc |
| | lamda_mult = (out_features1+out_features2)*1.0 / (self.kwargs['inc_cls_num']) |
| | |
| | if task_idx > 0: |
| | self.cur_lamda = self.kwargs['lamda'] * math.sqrt(lamda_mult) |
| | else: |
| | self.cur_lamda = self.kwargs['lamda'] |
| |
|
| | self._init_new_fc(task_idx, buffer, train_loader) |
| |
|
| | if task_idx == 0: |
| | self.loss_fn = nn.CrossEntropyLoss() |
| | else: |
| | self.loss_fn1 = nn.CosineEmbeddingLoss() |
| | self.loss_fn2 = nn.CrossEntropyLoss() |
| | self.loss_fn3 = nn.MarginRankingLoss(margin=self.kwargs['dist']) |
| |
|
| | self.ref_model.eval() |
| | self.num_old_classes = self.ref_model.classifier.out_features |
| | self.handle_ref_features = self.ref_model.classifier.register_forward_hook(get_ref_features) |
| | self.handle_cur_features = self.network.classifier.register_forward_hook(get_cur_features) |
| | self.handle_old_scores_bs = self.network.classifier.fc1.register_forward_hook(get_old_scores_before_scale) |
| | self.handle_new_scores_bs = self.network.classifier.fc2.register_forward_hook(get_new_scores_before_scale) |
| |
|
| | self.network = self.network.to(self.device) |
| | if self.ref_model is not None: |
| | self.ref_model = self.ref_model.to(self.device) |
| |
|
| | def _init_new_fc(self, task_idx, buffer, train_loader): |
| | if task_idx == 0: |
| | return |
| | old_embedding_norm = self.network.classifier.fc1.weight.data.norm(dim=1, keepdim=True) |
| | average_old_embedding_norm = torch.mean(old_embedding_norm, dim=0).to('cpu').type(torch.DoubleTensor) |
| | feature_model = self.network.backbone |
| | num_features = self.network.classifier.in_features |
| | novel_embedding = torch.zeros((self.kwargs['inc_cls_num'], num_features)) |
| |
|
| | tmp_datasets = copy.deepcopy(train_loader.dataset) |
| | for cls_idx in range(self.network.classifier.fc1.out_features, self.network.classifier.fc1.out_features + self.network.classifier.fc2.out_features): |
| | cls_dataset = train_loader.dataset |
| | task_data, task_target = cls_dataset.images, cls_dataset.labels |
| | cls_indices = np.where(np.array(task_target) == cls_idx) |
| | cls_data, cls_target = np.array([task_data[i] for i in cls_indices[0]]), np.array([task_target[i] for i in cls_indices[0]]) |
| | tmp_datasets.images = cls_data |
| | tmp_datasets.labels = cls_target |
| | tmp_loader = DataLoader(tmp_datasets, batch_size=128, shuffle=False, num_workers=2) |
| | num_samples = cls_data.shape[0] |
| | cls_features = self._compute_feature(feature_model, tmp_loader, num_samples, num_features) |
| | norm_features = F.normalize(torch.from_numpy(cls_features), p=2, dim=1) |
| | cls_embedding = torch.mean(norm_features, dim=0) |
| | novel_embedding[cls_idx-self.network.classifier.fc1.out_features] = F.normalize(cls_embedding, p=2, dim=0) * average_old_embedding_norm |
| | |
| | self.network.to(self.device) |
| | self.network.classifier.fc2.weight.data = novel_embedding.to(self.device) |
| |
|
| | def _compute_feature(self, feature_model, loader, num_samples, num_features): |
| | feature_model.eval() |
| | features = np.zeros([num_samples, num_features]) |
| | start_idx = 0 |
| | with torch.no_grad(): |
| | for batch_idx, batch in enumerate(loader): |
| | inputs, labels = batch['image'], batch['label'] |
| | inputs = inputs.to(self.device) |
| | features[start_idx:start_idx+inputs.shape[0], :] = np.squeeze(feature_model.feature(inputs).cpu()) |
| | start_idx = start_idx+inputs.shape[0] |
| | assert(start_idx==num_samples) |
| | return features |
| |
|
| |
|
| | def observe(self, data): |
| | x, y = data['image'], data['label'] |
| | x = x.to(self.device) |
| | y = y.to(self.device) |
| | logit = self.network(x) |
| |
|
| | if self.task_idx == 0: |
| | loss = self.loss_fn(logit, y) |
| | else: |
| | ref_outputs = self.ref_model(x) |
| | loss = self.loss_fn1(cur_features, ref_features.detach(), \ |
| | torch.ones(x.size(0)).to(self.device)) * self.cur_lamda |
| | |
| | loss += self.loss_fn2(logit, y) |
| | |
| | outputs_bs = torch.cat((old_scores, new_scores), dim=1) |
| | assert(outputs_bs.size()==logit.size()) |
| | gt_index = torch.zeros(outputs_bs.size()).to(self.device) |
| | gt_index = gt_index.scatter(1, y.view(-1,1), 1).ge(0.5) |
| | gt_scores = outputs_bs.masked_select(gt_index) |
| | max_novel_scores = outputs_bs[:, self.num_old_classes:].topk(self.K, dim=1)[0] |
| | hard_index = y.lt(self.num_old_classes) |
| | hard_num = torch.nonzero(hard_index).size(0) |
| | |
| | if hard_num > 0: |
| | gt_scores = gt_scores[hard_index].view(-1, 1).repeat(1, self.K) |
| | max_novel_scores = max_novel_scores[hard_index] |
| | assert(gt_scores.size() == max_novel_scores.size()) |
| | assert(gt_scores.size(0) == hard_num) |
| | loss += self.loss_fn3(gt_scores.view(-1, 1), \ |
| | max_novel_scores.view(-1, 1), torch.ones(hard_num*self.K, 1).to(self.device)) * self.lw_mr |
| |
|
| | pred = torch.argmax(logit, dim=1) |
| | acc = torch.sum(pred == y).item() |
| |
|
| | return pred, acc / x.size(0), loss |
| |
|
| | def after_task(self, task_idx, buffer, train_loader, test_loaders): |
| | if self.task_idx > 0: |
| | self.handle_ref_features.remove() |
| | self.handle_cur_features.remove() |
| | self.handle_old_scores_bs.remove() |
| | self.handle_new_scores_bs.remove() |
| |
|
| | def inference(self, data): |
| |
|
| | x, y = data['image'].to(self.device), data['label'].to(self.device) |
| |
|
| | logit = self.network(x) |
| | pred = torch.argmax(logit, dim=1) |
| |
|
| | acc = torch.sum(pred == y).item() |
| | return pred, acc / x.size(0) |
| |
|
| | def get_parameters(self, config): |
| | if self.task_idx > 0: |
| | |
| | ignored_params = list(map(id, self.network.classifier.fc1.parameters())) |
| | base_params = filter(lambda p: id(p) not in ignored_params, \ |
| | self.network.parameters()) |
| | tg_params =[{'params': base_params, 'lr': 0.1, 'weight_decay': 5e-4}, \ |
| | {'params': self.network.classifier.fc1.parameters(), 'lr': 0, 'weight_decay': 0}] |
| | else: |
| | tg_params = self.network.parameters() |
| | |
| | return tg_params |
