Enhanced-U-Net/models/modules.py

import torch
from torch import nn
from torch.nn import functional as F
from einops import rearrange, repeat
import math

""" Local Context Attention Module"""

def scaled_dot_product(q, k, v, mask=None):
    d_k = q.size()[-1]
    attn_logits = torch.matmul(q, k.transpose(-2, -1))
    attn_logits = attn_logits / math.sqrt(d_k)
    if mask is not None:
        attn_logits = attn_logits.masked_fill(mask == 0, -9e15)
    attention = F.softmax(attn_logits, dim=-1)
    values = torch.matmul(attention, v)
    return values, attention


class LCA(nn.Module):
    def __init__(self):
        super(LCA, self).__init__()

    def forward(self, x, pred):
        residual = x
        out = residual

        return out


""" Global Context Module"""


class GCM(nn.Module):
    def __init__(self, in_channels, out_channels):
        super(GCM, self).__init__()
        pool_size = [1, 3, 5]
        out_channel_list = [256, 128, 64, 64]
        upsampe_scale = [2, 4, 8, 16]
        GClist = []
        GCoutlist = []
        for ps in pool_size:
            GClist.append(nn.Sequential(
                nn.AdaptiveAvgPool2d(ps),
                nn.Conv2d(in_channels, out_channels, 1, 1),
                nn.ReLU(inplace=True)))
        GClist.append(nn.Sequential(
            nn.Conv2d(in_channels, out_channels, 1, 1),
            nn.ReLU(inplace=True),
            NonLocalBlock(out_channels)))
        self.GCmodule = nn.ModuleList(GClist)
        for i in range(4):
            GCoutlist.append(nn.Sequential(nn.Conv2d(out_channels * 4, out_channel_list[i], 3, 1, 1),
                                           nn.ReLU(inplace=True),
                                           nn.Upsample(scale_factor=upsampe_scale[i], mode='bilinear')))
        self.GCoutmodel = nn.ModuleList(GCoutlist)

    def forward(self, x):
        xsize = x.size()[2:]
        global_context = []
        for i in range(len(self.GCmodule) - 1):
            global_context.append(F.interpolate(self.GCmodule[i](x), xsize, mode='bilinear', align_corners=True))
        global_context.append(self.GCmodule[-1](x))
        global_context = torch.cat(global_context, dim=1)

        output = []
        for i in range(len(self.GCoutmodel)):
            output.append(self.GCoutmodel[i](global_context))

        return output


""" Adaptive Selection Module"""


class ASM(nn.Module):
    def __init__(self, in_channels, all_channels):
        super(ASM, self).__init__()
        self.non_local = NonLocalBlock(in_channels)

    def forward(self, lc, fuse, gc):
        fuse = self.non_local(fuse)
        fuse = torch.cat([lc, fuse, gc], dim=1)

        return fuse


"""
Squeeze and Excitation Layer

https://arxiv.org/abs/1709.01507

"""


class SELayer(nn.Module):
    def __init__(self, channel, reduction=16):
        super(SELayer, self).__init__()
        self.avg_pool = nn.AdaptiveAvgPool2d(1)
        self.fc = nn.Sequential(
            nn.Linear(channel, channel // reduction, bias=False),
            nn.ReLU(inplace=True),
            nn.Linear(channel // reduction, channel, bias=False),
            nn.Sigmoid()
        )

    def forward(self, x):
        b, c, _, _ = x.size()
        y = self.avg_pool(x).view(b, c)
        y = self.fc(y).view(b, c, 1, 1)
        return x * y.expand_as(x)


"""
Non Local Block

https://arxiv.org/abs/1711.07971
"""


class NonLocalBlock(nn.Module):
    def __init__(self, in_channels, inter_channels=None, sub_sample=True, bn_layer=True):
        super(NonLocalBlock, self).__init__()

        self.sub_sample = sub_sample

        self.in_channels = in_channels
        self.inter_channels = inter_channels

        if self.inter_channels is None:
            self.inter_channels = in_channels // 2
            if self.inter_channels == 0:
                self.inter_channels = 1

        self.g = nn.Conv2d(in_channels=self.in_channels, out_channels=self.inter_channels,
                           kernel_size=1, stride=1, padding=0)

        if bn_layer:
            self.W = nn.Sequential(
                nn.Conv2d(in_channels=self.inter_channels, out_channels=self.in_channels,
                          kernel_size=1, stride=1, padding=0),
                nn.BatchNorm2d(self.in_channels)
            )
            nn.init.constant_(self.W[1].weight, 0)
            nn.init.constant_(self.W[1].bias, 0)
        else:
            self.W = nn.Conv2d(in_channels=self.inter_channels, out_channels=self.in_channels,
                               kernel_size=1, stride=1, padding=0)
            nn.init.constant_(self.W.weight, 0)
            nn.init.constant_(self.W.bias, 0)

        self.theta = nn.Conv2d(in_channels=self.in_channels, out_channels=self.inter_channels,
                               kernel_size=1, stride=1, padding=0)
        self.phi = nn.Conv2d(in_channels=self.in_channels, out_channels=self.inter_channels,
                             kernel_size=1, stride=1, padding=0)

        if sub_sample:
            self.g = nn.Sequential(self.g, nn.MaxPool2d(kernel_size=(2, 2)))
            self.phi = nn.Sequential(self.phi, nn.MaxPool2d(kernel_size=(2, 2)))

    def forward(self, x):

        batch_size = x.size(0)

        g_x = self.g(x).view(batch_size, self.inter_channels, -1)
        g_x = g_x.permute(0, 2, 1)

        theta_x = self.theta(x).view(batch_size, self.inter_channels, -1)
        theta_x = theta_x.permute(0, 2, 1)
        phi_x = self.phi(x).view(batch_size, self.inter_channels, -1)
        f = torch.matmul(theta_x, phi_x)
        f_div_C = F.softmax(f, dim=-1)

        y = torch.matmul(f_div_C, g_x)
        y = y.permute(0, 2, 1).contiguous()
        y = y.view(batch_size, self.inter_channels, *x.size()[2:])
        W_y = self.W(y)
        z = W_y + x

        return z


#AGCM Module
class CrossNonLocalBlock(nn.Module):
    def __init__(self, in_channels_source,in_channels_target, inter_channels, sub_sample=False, bn_layer=True):
        super(CrossNonLocalBlock, self).__init__()

        self.sub_sample = sub_sample

        self.in_channels_source = in_channels_source
        self.in_channels_target = in_channels_target
        self.inter_channels = inter_channels

        """
        if self.inter_channels is None:
            self.inter_channels = in_channels // 2
            if self.inter_channels == 0:
                self.inter_channels = 1
        """
        self.g = nn.Conv2d(in_channels=self.in_channels_source, out_channels=self.inter_channels,
                           kernel_size=1, stride=1, padding=0)
        self.theta = nn.Conv2d(in_channels=self.in_channels_source, out_channels=self.inter_channels,
                               kernel_size=1, stride=1, padding=0)
        self.phi = nn.Conv2d(in_channels=self.in_channels_target, out_channels=self.inter_channels,
                             kernel_size=1, stride=1, padding=0)

        if bn_layer:
            self.W = nn.Sequential(
                nn.Conv2d(in_channels=self.inter_channels, out_channels=self.in_channels_target,
                          kernel_size=1, stride=1, padding=0),
                nn.BatchNorm2d(self.in_channels_target)
            )
            nn.init.constant_(self.W[1].weight, 0)
            nn.init.constant_(self.W[1].bias, 0)
     
        if sub_sample:
            self.g = nn.Sequential(self.g, nn.MaxPool2d(kernel_size=(2, 2)))
            self.phi = nn.Sequential(self.phi, nn.MaxPool2d(kernel_size=(2, 2)))

    def forward(self,x,l):

        batch_size = x.size(0)
        g_x = self.g(x).view(batch_size, self.inter_channels, -1)
        g_x = g_x.permute(0, 2, 1) #source
        theta_x1 = self.theta(x)
        theta_x = self.theta(x).view(batch_size, self.inter_channels, -1)
        theta_x = theta_x.permute(0, 2, 1) #source
        phi_x = self.phi(l).view(batch_size, self.inter_channels, -1) #target
        f = torch.matmul(theta_x, phi_x) 
        f_div_C = F.softmax(f, dim=-1)
        f_div_C = f_div_C.permute(0,2,1)
        y = torch.matmul(f_div_C, g_x)     
        y = y.permute(0, 2, 1).contiguous()
        y = y.view(batch_size, self.inter_channels, *l.size()[2:])
        W_y = self.W(y)
        z = W_y + l

        return z



#SFEM module
class NonLocalBlock_PatchWise(nn.Module):

    def __init__(self, in_channel, inter_channel, patch_factor):
        super(NonLocalBlock_PatchWise, self).__init__()
        "Embedding dimension must be 0 modulo number of heads."
        self.in_channel = in_channel
        self.patch_factor = patch_factor
        self.patch_width = int(8/self.patch_factor)
        self.patch_height = int(8/self.patch_factor)
        self.stride_width = int(8/self.patch_factor)
        self.stride_height = int(8/self.patch_factor)
        self.unfold = nn.Unfold(kernel_size=(self.patch_width, self.patch_height), stride=(self.stride_width, self.stride_height)) 
        

        self.adp = nn.AdaptiveAvgPool2d(8)
        self.bottleneck = nn.Conv2d(64,inter_channel,kernel_size=(1,1))
        self.non_block =  NonLocalBlock(self.in_channel)
        self.adp_post = nn.AdaptiveAvgPool2d((8,8))


    def forward(self, x):
        batch_size = x.size(0)
        x_up = self.adp(x)
        x_up = self.unfold(x)
        batch_size,p_dim,p_size = x_up.size()
        x_up = x_up.view(batch_size,-1,self.in_channel,p_size)
        final_output = torch.tensor([]).cuda()
        index = torch.arange(0,p_size,1,dtype=torch.int64).cuda()
        for i in range(int(p_size)):
            divide = torch.index_select(x_up, 3, index[i])
            divide = divide.view(batch_size,-1,self.in_channel)
            patch_width = int(divide.size(1) ** 0.5)
            divide = divide.reshape(batch_size,self.in_channel,patch_width,patch_width) # tensor to operate on
            attn = self.non_block(divide)
            output = attn.view(batch_size,-1,self.in_channel,1)
            final_output = torch.cat((final_output,output),dim=3)
            
        
    
        final_output = final_output.view(batch_size, self.in_channel, 8,8) 
        
            
        return final_output
       

class GCM_up(nn.Module):
    def __init__(self, in_channels, out_channels):
        super(GCM_up, self).__init__()
        self.adp = nn.AdaptiveAvgPool2d((8,8))
        self.patch1 = NonLocalBlock_PatchWise(in_channels,out_channels,2)
        self.patch2 = NonLocalBlock_PatchWise(in_channels,out_channels,4)
        self.patch3 = NonLocalBlock(256,64)
        self.fuse = SELayer(3*256)
        self.conv = nn.Conv2d(3*256, out_channels, 1, 1)
        self.relu = nn.ReLU(inplace=True)
          

    def forward(self, x):
       
        b,c,h,w = x.size()
        x = self.adp(x)
        patch1 = self.patch1(x)
        patch2 = self.patch2(x)
        patch3 = self.patch3(x)
        global_cat = torch.cat((patch1, patch2, patch3), dim=1)
        fuse = self.relu(self.conv(self.fuse(global_cat)))
        adp_post = nn.AdaptiveAvgPool2d((h,w))
        fuse = adp_post(fuse)
        return fuse