models/GSSN.py

import torch
import torch.nn as nn
import torch.nn.functional as F
from torchvision import utils
from collections import OrderedDict
import numpy as np
import matplotlib.cm as cm
import matplotlib as mpl

from .abs_model import abs_model
from .blocks import *
from .SSN import SSN
from .SSN_v1 import SSN_v1
from .Loss.Loss import norm_loss


class GSSN(abs_model):
    def __init__(self, opt):
        mid_act      = opt['model']['mid_act']
        out_act      = opt['model']['out_act']
        in_channels  = opt['model']['in_channels']
        out_channels = opt['model']['out_channels']
        resnet       = opt['model']['resnet']
        self.ncols   = opt['hyper_params']['n_cols']
        self.focal   = opt['model']['focal']

        if 'backbone' not in opt['model'].keys():
            self.model = SSN(in_channels=in_channels,
                             out_channels=out_channels,
                             mid_act=mid_act,
                             out_act=out_act,
                             resnet=resnet)

        else:
            backbone = opt['model']['backbone']
            if backbone == 'vanilla':
                self.model = SSN(in_channels=in_channels,
                                 out_channels=out_channels,
                                 mid_act=mid_act,
                                 out_act=out_act,
                                 resnet=resnet)
            elif backbone == 'SSN_v1':
                self.model = SSN_v1(in_channels=in_channels,
                                    out_channels=out_channels,
                                    mid_act=mid_act,
                                    out_act=out_act,
                                    resnet=resnet)
            else:
                raise NotImplementedError('{} has not implemented yet'.format(backbone))


        self.optimizer = get_optimizer(opt, self.model)
        self.visualization = {}

        self.norm_loss = norm_loss()

        # inference related
        BINs    = 100
        MAX_RAD = 20
        self.size_interval     = MAX_RAD / BINs
        self.soft_distribution = [[np.exp(-0.2 * (i - j) ** 2) for i in np.arange(BINs)] for j in np.arange(BINs)]

    def setup_input(self, x):
        return x


    def forward(self, x):
        x, softness = x
        return self.model(x, softness)


    def compute_loss(self, y, pred):
        b = y.shape[0]

        total_loss = self.norm_loss.loss(y, pred)

        if self.focal:
            total_loss = torch.pow(total_loss, 3)

        return total_loss


    def supervise(self, input_x, y, is_training:bool)->float:
        optimizer = self.optimizer
        model = self.model

        x, softness = input_x['x'], input_x['softness']

        optimizer.zero_grad()
        pred = model(x, softness)
        loss = self.compute_loss(y, pred)

        if is_training:
            loss.backward()
            optimizer.step()

        xc = x.shape[1]
        for i in range(xc):
            self.visualization['x{}'.format(i)] = x[:, i:i+1].detach()

        self.visualization['y']    = y.detach()
        self.visualization['pred'] = pred.detach()

        return loss.item()


    def get_visualize(self) -> OrderedDict:
        """ Convert to visualization numpy array
        """
        nrows          = self.ncols
        visualizations = self.visualization
        ret_vis        = OrderedDict()

        for k, v in visualizations.items():
            batch = v.shape[0]
            n     = min(nrows, batch)

            plot_v = v[:n]
            ret_vis[k] = np.clip(utils.make_grid(plot_v.cpu(), nrow=nrows).numpy().transpose(1,2,0), 0.0, 1.0)
            ret_vis[k] = self.plasma(ret_vis[k])

        return ret_vis


    def get_logs(self):
        pass


    def inference(self, x):
        x, l, device = x['x'], x['l'], x['device']

        x = torch.from_numpy(x.transpose((2,0,1))).unsqueeze(dim=0).to(device)
        l = torch.from_numpy(np.array(self.soft_distribution[int(l/self.size_interval)]).astype(np.float32)).unsqueeze(dim=0).to(device)

        pred = self.forward((x, l))
        pred = pred[0].detach().cpu().numpy().transpose((1,2,0))
        return pred


    def batch_inference(self, x):
        x, l = x['x'], x['softness']
        pred = self.forward((x, l))
        return pred


    """ Getter & Setter
    """
    def get_models(self) -> dict:
        return {'model': self.model}


    def get_optimizers(self) -> dict:
        return {'optimizer': self.optimizer}


    def set_models(self, models: dict) :
        # input test
        if 'model' not in models.keys():
            raise ValueError('{} not in self.model'.format('model'))

        self.model = models['model']


    def set_optimizers(self, optimizer: dict):
        self.optimizer = optimizer['optimizer']


    ####################
    # Personal Methods #
    ####################
    def plasma(self, x):
        norm   = mpl.colors.Normalize(vmin=0.0, vmax=1)
        mapper = cm.ScalarMappable(norm=norm, cmap='plasma')
        bimg   = mapper.to_rgba(x[:,:,0])[:,:,:3]

        return bimg