<filename>self_supervised/vision/simclr.py
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/10 - simclr.ipynb (unless otherwise specified).

__all__ = ['SimCLRModel', 'create_simclr_model', 'get_simclr_aug_pipelines', 'SimCLR', 'DistributedSimCLR']

# Cell
from fastai.vision.all import *
from ..augmentations import *
from ..layers import *

# Cell
class SimCLRModel(Module):
    "Compute predictions of concatenated xi and xj"
    def __init__(self,encoder,projector): self.encoder,self.projector = encoder,projector
    def forward(self,x): return self.projector(self.encoder(x))

# Cell
def create_simclr_model(encoder, hidden_size=256, projection_size=128, bn=False, nlayers=2):
    "Create SimCLR model"
    n_in  = in_channels(encoder)
    with torch.no_grad(): representation = encoder(torch.randn((2,n_in,128,128)))
    projector = create_mlp_module(representation.size(1), hidden_size, projection_size, bn=bn, nlayers=nlayers)
    apply_init(projector)
    return SimCLRModel(encoder, projector)

# Cell
@delegates(get_multi_aug_pipelines)
def get_simclr_aug_pipelines(size, **kwargs): return get_multi_aug_pipelines(n=2, size=size, **kwargs)

# Cell
class SimCLR(Callback):
    order,run_valid = 9,True
    def __init__(self, aug_pipelines, temp=0.07, print_augs=False):
        assert_aug_pipelines(aug_pipelines)
        self.aug1, self.aug2 = aug_pipelines
        if print_augs: print(self.aug1), print(self.aug2)
        store_attr('temp')


    def before_fit(self):
        self.learn.loss_func = self.lf


    def before_batch(self):
        xi,xj = self.aug1(self.x), self.aug2(self.x)
        self.learn.xb = (torch.cat([xi, xj]),)
        bs = self.learn.xb[0].shape[0]
        self.learn.yb = (torch.arange(bs, device=self.dls.device).roll(bs//2),)


    def _remove_diag(self, x):
        bs = x.shape[0]
        return x[~torch.eye(bs).bool()].reshape(bs,bs-1)


    def lf(self, pred, *yb):
        pred, targ = F.normalize(pred, dim=1), yb[0]
        sim = self._remove_diag(pred @ pred.T) / self.temp
        targ = self._remove_diag(torch.eye(targ.shape[0], device=self.dls.device)[targ]).nonzero()[:,-1]
        return F.cross_entropy(sim, targ)


    @torch.no_grad()
    def show(self, n=1):
        bs = self.learn.x.size(0)//2
        x1,x2  = self.learn.x[:bs], self.learn.x[bs:]
        idxs = np.random.choice(range(bs),n,False)
        x1 = self.aug1.decode(x1[idxs].to('cpu').clone()).clamp(0,1)
        x2 = self.aug2.decode(x2[idxs].to('cpu').clone()).clamp(0,1)
        images = []
        for i in range(n): images += [x1[i],x2[i]]
        return show_batch(x1[0], None, images, max_n=len(images), nrows=n)

# Cell
from ..dist import GatherLayer

class DistributedSimCLR(Callback):
    order,run_valid = 9,True
    def __init__(self, aug_pipelines=[], temp=0.07, print_augs=False):
        assert_aug_pipelines(aug_pipelines)
        self.aug1, self.aug2 = aug_pipelines
        if print_augs: print(self.aug1), print(self.aug2)
        store_attr('temp')


    def before_fit(self):
        self.learn.loss_func = self.lf


    def before_batch(self):
        xi,xj = self.aug1(self.x), self.aug2(self.x)
        self.learn.xb = (torch.cat([xi, xj]),)
        bs = self.learn.xb[0].shape[0]
        self.learn.yb = (torch.arange(bs, device=self.dls.device).roll(bs//2),)


    def _remove_diag(self, x):
        bs = x.shape[0]
        return x[~torch.eye(bs).bool()].reshape(bs,bs-1)


    def lf(self, pred, *yb):
        # gather and reorder
        all_preds = list(GatherLayer.apply(pred)); all_preds.pop(rank_distrib())
        all_preds = torch.cat([pred]+all_preds)

        pred, all_preds, targ = F.normalize(pred, dim=1), F.normalize(all_preds, dim=1), yb[0]
        sim = self._remove_diag(pred @ all_preds.T) / self.temp
        targ = self._remove_diag(torch.eye(targ.shape[0], device=self.dls.device)[targ]).nonzero()[:,-1]
        return F.cross_entropy(sim, targ)
<filename>self_supervised/vision/simclr.py
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/10 - simclr.ipynb (unless otherwise specified).

__all__ = ['SimCLRModel', 'create_simclr_model', 'get_simclr_aug_pipelines', 'SimCLR', 'DistributedSimCLR']

# Cell
from fastai.vision.all import *
from ..augmentations import *
from ..layers import *

# Cell
class SimCLRModel(Module):
    "Compute predictions of concatenated xi and xj"
    def __init__(self,encoder,projector): self.encoder,self.projector = encoder,projector
    def forward(self,x): return self.projector(self.encoder(x))

# Cell
def create_simclr_model(encoder, hidden_size=256, projection_size=128, bn=False, nlayers=2):
    "Create SimCLR model"
    n_in  = in_channels(encoder)
    with torch.no_grad(): representation = encoder(torch.randn((2,n_in,128,128)))
    projector = create_mlp_module(representation.size(1), hidden_size, projection_size, bn=bn, nlayers=nlayers)
    apply_init(projector)
    return SimCLRModel(encoder, projector)

# Cell
@delegates(get_multi_aug_pipelines)
def get_simclr_aug_pipelines(size, **kwargs): return get_multi_aug_pipelines(n=2, size=size, **kwargs)

# Cell
class SimCLR(Callback):
    order,run_valid = 9,True
    def __init__(self, aug_pipelines, temp=0.07, print_augs=False):
        assert_aug_pipelines(aug_pipelines)
        self.aug1, self.aug2 = aug_pipelines
        if print_augs: print(self.aug1), print(self.aug2)
        store_attr('temp')


    def before_fit(self):
        self.learn.loss_func = self.lf


    def before_batch(self):
        xi,xj = self.aug1(self.x), self.aug2(self.x)
        self.learn.xb = (torch.cat([xi, xj]),)
        bs = self.learn.xb[0].shape[0]
        self.learn.yb = (torch.arange(bs, device=self.dls.device).roll(bs//2),)


    def _remove_diag(self, x):
        bs = x.shape[0]
        return x[~torch.eye(bs).bool()].reshape(bs,bs-1)


    def lf(self, pred, *yb):
        pred, targ = F.normalize(pred, dim=1), yb[0]
        sim = self._remove_diag(pred @ pred.T) / self.temp
        targ = self._remove_diag(torch.eye(targ.shape[0], device=self.dls.device)[targ]).nonzero()[:,-1]
        return F.cross_entropy(sim, targ)


    @torch.no_grad()
    def show(self, n=1):
        bs = self.learn.x.size(0)//2
        x1,x2  = self.learn.x[:bs], self.learn.x[bs:]
        idxs = np.random.choice(range(bs),n,False)
        x1 = self.aug1.decode(x1[idxs].to('cpu').clone()).clamp(0,1)
        x2 = self.aug2.decode(x2[idxs].to('cpu').clone()).clamp(0,1)
        images = []
        for i in range(n): images += [x1[i],x2[i]]
        return show_batch(x1[0], None, images, max_n=len(images), nrows=n)

# Cell
from ..dist import GatherLayer

class DistributedSimCLR(Callback):
    order,run_valid = 9,True
    def __init__(self, aug_pipelines=[], temp=0.07, print_augs=False):
        assert_aug_pipelines(aug_pipelines)
        self.aug1, self.aug2 = aug_pipelines
        if print_augs: print(self.aug1), print(self.aug2)
        store_attr('temp')


    def before_fit(self):
        self.learn.loss_func = self.lf


    def before_batch(self):
        xi,xj = self.aug1(self.x), self.aug2(self.x)
        self.learn.xb = (torch.cat([xi, xj]),)
        bs = self.learn.xb[0].shape[0]
        self.learn.yb = (torch.arange(bs, device=self.dls.device).roll(bs//2),)


    def _remove_diag(self, x):
        bs = x.shape[0]
        return x[~torch.eye(bs).bool()].reshape(bs,bs-1)


    def lf(self, pred, *yb):
        # gather and reorder
        all_preds = list(GatherLayer.apply(pred)); all_preds.pop(rank_distrib())
        all_preds = torch.cat([pred]+all_preds)

        pred, all_preds, targ = F.normalize(pred, dim=1), F.normalize(all_preds, dim=1), yb[0]
        sim = self._remove_diag(pred @ all_preds.T) / self.temp
        targ = self._remove_diag(torch.eye(targ.shape[0], device=self.dls.device)[targ]).nonzero()[:,-1]
        return F.cross_entropy(sim, targ)
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/10 - simclr.ipynb (unless otherwise specified). # Cell # Cell # Cell # Cell # Cell # Cell # gather and reorder