import os import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import argparse import copy from typing import Optional import numpy as np import torch import torch.nn.functional as F from torch import Tensor, nn from .extractor import BasicEncoder from .position_encoding import build_position_encoding class attnLayer(nn.Module): def __init__( self, d_model, nhead=8, dim_feedforward=2048, dropout=0.1, activation="relu", normalize_before=False, ): super().__init__() self.self_attn = nn.MultiheadAttention(d_model, nhead, dropout=dropout) self.multihead_attn_list = nn.ModuleList( [ copy.deepcopy(nn.MultiheadAttention(d_model, nhead, dropout=dropout)) for i in range(2) ] ) # Implementation of Feedforward model self.linear1 = nn.Linear(d_model, dim_feedforward) self.dropout = nn.Dropout(dropout) self.linear2 = nn.Linear(dim_feedforward, d_model) self.norm1 = nn.LayerNorm(d_model) self.norm2_list = nn.ModuleList( [copy.deepcopy(nn.LayerNorm(d_model)) for i in range(2)] ) self.norm3 = nn.LayerNorm(d_model) self.dropout1 = nn.Dropout(dropout) self.dropout2_list = nn.ModuleList( [copy.deepcopy(nn.Dropout(dropout)) for i in range(2)] ) self.dropout3 = nn.Dropout(dropout) self.activation = _get_activation_fn(activation) self.normalize_before = normalize_before def with_pos_embed(self, tensor, pos: Optional[Tensor]): return tensor if pos is None else tensor + pos def forward_post( self, tgt, memory_list, tgt_mask=None, memory_mask=None, tgt_key_padding_mask=None, memory_key_padding_mask=None, pos=None, memory_pos=None, ): q = k = self.with_pos_embed(tgt, pos) tgt2 = self.self_attn( q, k, value=tgt, attn_mask=tgt_mask, key_padding_mask=tgt_key_padding_mask )[0] tgt = tgt + self.dropout1(tgt2) tgt = self.norm1(tgt) for memory, multihead_attn, norm2, dropout2, m_pos in zip( memory_list, self.multihead_attn_list, self.norm2_list, self.dropout2_list, memory_pos, ): tgt2 = multihead_attn( query=self.with_pos_embed(tgt, pos), key=self.with_pos_embed(memory, m_pos), value=memory, attn_mask=memory_mask, key_padding_mask=memory_key_padding_mask, )[0] tgt = tgt + dropout2(tgt2) tgt = norm2(tgt) tgt2 = self.linear2(self.dropout(self.activation(self.linear1(tgt)))) tgt = tgt + self.dropout3(tgt2) tgt = self.norm3(tgt) return tgt def forward_pre( self, tgt, memory, tgt_mask=None, memory_mask=None, tgt_key_padding_mask=None, memory_key_padding_mask=None, pos=None, memory_pos=None, ): tgt2 = self.norm1(tgt) q = k = self.with_pos_embed(tgt2, pos) tgt2 = self.self_attn( q, k, value=tgt2, attn_mask=tgt_mask, key_padding_mask=tgt_key_padding_mask )[0] tgt = tgt + self.dropout1(tgt2) tgt2 = self.norm2(tgt) tgt2 = self.multihead_attn( query=self.with_pos_embed(tgt2, pos), key=self.with_pos_embed(memory, memory_pos), value=memory, attn_mask=memory_mask, key_padding_mask=memory_key_padding_mask, )[0] tgt = tgt + self.dropout2(tgt2) tgt2 = self.norm3(tgt) tgt2 = self.linear2(self.dropout(self.activation(self.linear1(tgt2)))) tgt = tgt + self.dropout3(tgt2) return tgt def forward( self, tgt, memory_list, tgt_mask=None, memory_mask=None, tgt_key_padding_mask=None, memory_key_padding_mask=None, pos=None, memory_pos=None, ): if self.normalize_before: return self.forward_pre( tgt, memory_list, tgt_mask, memory_mask, tgt_key_padding_mask, memory_key_padding_mask, pos, memory_pos, ) return self.forward_post( tgt, memory_list, tgt_mask, memory_mask, tgt_key_padding_mask, memory_key_padding_mask, pos, memory_pos, ) def _get_clones(module, N): return nn.ModuleList([copy.deepcopy(module) for i in range(N)]) def _get_activation_fn(activation): """Return an activation function given a string""" if activation == "relu": return F.relu if activation == "gelu": return F.gelu if activation == "glu": return F.glu raise RuntimeError(f"activation should be relu/gelu, not {activation}.") class TransDecoder(nn.Module): def __init__(self, num_attn_layers, hidden_dim=128): super(TransDecoder, self).__init__() attn_layer = attnLayer(hidden_dim) self.layers = _get_clones(attn_layer, num_attn_layers) self.position_embedding = build_position_encoding(hidden_dim) def forward(self, imgf, query_embed): pos = self.position_embedding( torch.ones(imgf.shape[0], imgf.shape[2], imgf.shape[3]).bool().cuda() ) # torch.Size([1, 128, 36, 36]) bs, c, h, w = imgf.shape imgf = imgf.flatten(2).permute(2, 0, 1) # query_embed = query_embed.unsqueeze(1).repeat(1, bs, 1) pos = pos.flatten(2).permute(2, 0, 1) for layer in self.layers: query_embed = layer(query_embed, [imgf], pos=pos, memory_pos=[pos, pos]) query_embed = query_embed.permute(1, 2, 0).reshape(bs, c, h, w) return query_embed class TransEncoder(nn.Module): def __init__(self, num_attn_layers, hidden_dim=128): super(TransEncoder, self).__init__() attn_layer = attnLayer(hidden_dim) self.layers = _get_clones(attn_layer, num_attn_layers) self.position_embedding = build_position_encoding(hidden_dim) def forward(self, imgf): pos = self.position_embedding( torch.ones(imgf.shape[0], imgf.shape[2], imgf.shape[3]).bool().cuda() ) # torch.Size([1, 128, 36, 36]) bs, c, h, w = imgf.shape imgf = imgf.flatten(2).permute(2, 0, 1) pos = pos.flatten(2).permute(2, 0, 1) for layer in self.layers: imgf = layer(imgf, [imgf], pos=pos, memory_pos=[pos, pos]) imgf = imgf.permute(1, 2, 0).reshape(bs, c, h, w) return imgf class FlowHead(nn.Module): def __init__(self, input_dim=128, hidden_dim=256): super(FlowHead, self).__init__() self.conv1 = nn.Conv2d(input_dim, hidden_dim, 3, padding=1) self.conv2 = nn.Conv2d(hidden_dim, 2, 3, padding=1) self.relu = nn.ReLU(inplace=True) def forward(self, x): return self.conv2(self.relu(self.conv1(x))) class UpdateBlock(nn.Module): def __init__(self, hidden_dim=128): super(UpdateBlock, self).__init__() self.flow_head = FlowHead(hidden_dim, hidden_dim=256) self.mask = nn.Sequential( nn.Conv2d(hidden_dim, 256, 3, padding=1), nn.ReLU(inplace=True), nn.Conv2d(256, 64 * 9, 1, padding=0), ) def forward(self, imgf, coords1): mask = 0.25 * self.mask(imgf) # scale mask to balence gradients dflow = self.flow_head(imgf) coords1 = coords1 + dflow return mask, coords1 def coords_grid(batch, ht, wd): coords = torch.meshgrid(torch.arange(ht), torch.arange(wd)) coords = torch.stack(coords[::-1], dim=0).float() return coords[None].repeat(batch, 1, 1, 1) def upflow8(flow, mode="bilinear"): new_size = (8 * flow.shape[2], 8 * flow.shape[3]) return 8 * F.interpolate(flow, size=new_size, mode=mode, align_corners=True) class OverlapPatchEmbed(nn.Module): """Image to Patch Embedding""" def __init__(self, img_size=224, patch_size=7, stride=4, in_chans=3, embed_dim=768): super().__init__() img_size = to_2tuple(img_size) patch_size = to_2tuple(patch_size) self.img_size = img_size self.patch_size = patch_size self.H, self.W = img_size[0] // patch_size[0], img_size[1] // patch_size[1] self.num_patches = self.H * self.W self.proj = nn.Conv2d( in_chans, embed_dim, kernel_size=patch_size, stride=stride, padding=(patch_size[0] // 2, patch_size[1] // 2), ) self.norm = nn.LayerNorm(embed_dim) self.apply(self._init_weights) def _init_weights(self, m): if isinstance(m, nn.Linear): trunc_normal_(m.weight, std=0.02) if isinstance(m, nn.Linear) and m.bias is not None: nn.init.constant_(m.bias, 0) elif isinstance(m, nn.LayerNorm): nn.init.constant_(m.bias, 0) nn.init.constant_(m.weight, 1.0) elif isinstance(m, nn.Conv2d): fan_out = m.kernel_size[0] * m.kernel_size[1] * m.out_channels fan_out //= m.groups m.weight.data.normal_(0, math.sqrt(2.0 / fan_out)) if m.bias is not None: m.bias.data.zero_() def forward(self, x): x = self.proj(x) _, _, H, W = x.shape x = x.flatten(2).transpose(1, 2) x = self.norm(x) return x, H, W class GeoTr(nn.Module): def __init__(self): super(GeoTr, self).__init__() self.hidden_dim = hdim = 256 self.fnet = BasicEncoder(output_dim=hdim, norm_fn="instance") self.encoder_block = ["encoder_block" + str(i) for i in range(3)] for i in self.encoder_block: self.__setattr__(i, TransEncoder(2, hidden_dim=hdim)) self.down_layer = ["down_layer" + str(i) for i in range(2)] for i in self.down_layer: self.__setattr__(i, nn.Conv2d(256, 256, kernel_size=3, stride=2, padding=1)) self.decoder_block = ["decoder_block" + str(i) for i in range(3)] for i in self.decoder_block: self.__setattr__(i, TransDecoder(2, hidden_dim=hdim)) self.up_layer = ["up_layer" + str(i) for i in range(2)] for i in self.up_layer: self.__setattr__( i, nn.Upsample(scale_factor=2, mode="bilinear", align_corners=True) ) self.query_embed = nn.Embedding(81, self.hidden_dim) self.update_block = UpdateBlock(self.hidden_dim) def initialize_flow(self, img): N, C, H, W = img.shape coodslar = coords_grid(N, H, W).to(img.device) coords0 = coords_grid(N, H // 8, W // 8).to(img.device) coords1 = coords_grid(N, H // 8, W // 8).to(img.device) return coodslar, coords0, coords1 def upsample_flow(self, flow, mask): N, _, H, W = flow.shape mask = mask.view(N, 1, 9, 8, 8, H, W) mask = torch.softmax(mask, dim=2) up_flow = F.unfold(8 * flow, [3, 3], padding=1) up_flow = up_flow.view(N, 2, 9, 1, 1, H, W) up_flow = torch.sum(mask * up_flow, dim=2) up_flow = up_flow.permute(0, 1, 4, 2, 5, 3) return up_flow.reshape(N, 2, 8 * H, 8 * W) def forward(self, image1): fmap = self.fnet(image1) fmap = torch.relu(fmap) # fmap = self.TransEncoder(fmap) fmap1 = self.__getattr__(self.encoder_block[0])(fmap) fmap1d = self.__getattr__(self.down_layer[0])(fmap1) fmap2 = self.__getattr__(self.encoder_block[1])(fmap1d) fmap2d = self.__getattr__(self.down_layer[1])(fmap2) fmap3 = self.__getattr__(self.encoder_block[2])(fmap2d) query_embed0 = self.query_embed.weight.unsqueeze(1).repeat(1, fmap3.size(0), 1) fmap3d_ = self.__getattr__(self.decoder_block[0])(fmap3, query_embed0) fmap3du_ = ( self.__getattr__(self.up_layer[0])(fmap3d_).flatten(2).permute(2, 0, 1) ) fmap2d_ = self.__getattr__(self.decoder_block[1])(fmap2, fmap3du_) fmap2du_ = ( self.__getattr__(self.up_layer[1])(fmap2d_).flatten(2).permute(2, 0, 1) ) fmap_out = self.__getattr__(self.decoder_block[2])(fmap1, fmap2du_) # convex upsample baesd on fmap_out coodslar, coords0, coords1 = self.initialize_flow(image1) coords1 = coords1.detach() mask, coords1 = self.update_block(fmap_out, coords1) flow_up = self.upsample_flow(coords1 - coords0, mask) bm_up = coodslar + flow_up return bm_up ## upsample tensor 'src' to have the same spatial size with tensor 'tar' def _upsample_like(src, tar): src = F.interpolate(src, size=tar.shape[2:], mode="bilinear", align_corners=False) return src class REBNCONV(nn.Module): def __init__(self, in_ch=3, out_ch=3, dirate=1): super(REBNCONV, self).__init__() self.conv_s1 = nn.Conv2d( in_ch, out_ch, 3, padding=1 * dirate, dilation=1 * dirate ) self.bn_s1 = nn.BatchNorm2d(out_ch) self.relu_s1 = nn.ReLU(inplace=True) def forward(self, x): hx = x xout = self.relu_s1(self.bn_s1(self.conv_s1(hx))) return xout ### RSU-4 ### class RSU4(nn.Module): # UNet04DRES(nn.Module): def __init__(self, in_ch=3, mid_ch=12, out_ch=3): super(RSU4, self).__init__() self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1) self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1) self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1) self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=2) self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1) def forward(self, x): hx = x hxin = self.rebnconvin(hx) hx1 = self.rebnconv1(hxin) hx = self.pool1(hx1) hx2 = self.rebnconv2(hx) hx = self.pool2(hx2) hx3 = self.rebnconv3(hx) hx4 = self.rebnconv4(hx3) hx3d = self.rebnconv3d(torch.cat((hx4, hx3), 1)) hx3dup = _upsample_like(hx3d, hx2) hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1)) hx2dup = _upsample_like(hx2d, hx1) hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1)) return hx1d + hxin ### RSU-4F ### class RSU4F(nn.Module): # UNet04FRES(nn.Module): def __init__(self, in_ch=3, mid_ch=12, out_ch=3): super(RSU4F, self).__init__() self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1) self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1) self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=2) self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=4) self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=8) self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=4) self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=2) self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1) def forward(self, x): hx = x hxin = self.rebnconvin(hx) hx1 = self.rebnconv1(hxin) hx2 = self.rebnconv2(hx1) hx3 = self.rebnconv3(hx2) hx4 = self.rebnconv4(hx3) hx3d = self.rebnconv3d(torch.cat((hx4, hx3), 1)) hx2d = self.rebnconv2d(torch.cat((hx3d, hx2), 1)) hx1d = self.rebnconv1d(torch.cat((hx2d, hx1), 1)) return hx1d + hxin class sobel_net(nn.Module): def __init__(self): super().__init__() self.conv_opx = nn.Conv2d(1, 1, 3, bias=False) self.conv_opy = nn.Conv2d(1, 1, 3, bias=False) sobel_kernelx = np.array( [[-1, 0, 1], [-2, 0, 2], [-1, 0, 1]], dtype="float32" ).reshape((1, 1, 3, 3)) sobel_kernely = np.array( [[-1, -2, -1], [0, 0, 0], [1, 2, 1]], dtype="float32" ).reshape((1, 1, 3, 3)) self.conv_opx.weight.data = torch.from_numpy(sobel_kernelx) self.conv_opy.weight.data = torch.from_numpy(sobel_kernely) for p in self.parameters(): p.requires_grad = False def forward(self, im): # input rgb x = ( 0.299 * im[:, 0, :, :] + 0.587 * im[:, 1, :, :] + 0.114 * im[:, 2, :, :] ).unsqueeze( 1 ) # rgb2gray gradx = self.conv_opx(x) grady = self.conv_opy(x) x = (gradx**2 + grady**2) ** 0.5 x = (x - x.min()) / (x.max() - x.min()) x = F.pad(x, (1, 1, 1, 1)) x = torch.cat([im, x], dim=1) return x ##### U^2-Net #### class U2NET(nn.Module): def __init__(self, in_ch=3, out_ch=1): super(U2NET, self).__init__() self.edge = sobel_net() self.stage1 = RSU7(in_ch, 32, 64) self.pool12 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage2 = RSU6(64, 32, 128) self.pool23 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage3 = RSU5(128, 64, 256) self.pool34 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage4 = RSU4(256, 128, 512) self.pool45 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage5 = RSU4F(512, 256, 512) self.pool56 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage6 = RSU4F(512, 256, 512) # decoder self.stage5d = RSU4F(1024, 256, 512) self.stage4d = RSU4(1024, 128, 256) self.stage3d = RSU5(512, 64, 128) self.stage2d = RSU6(256, 32, 64) self.stage1d = RSU7(128, 16, 64) self.side1 = nn.Conv2d(64, out_ch, 3, padding=1) self.side2 = nn.Conv2d(64, out_ch, 3, padding=1) self.side3 = nn.Conv2d(128, out_ch, 3, padding=1) self.side4 = nn.Conv2d(256, out_ch, 3, padding=1) self.side5 = nn.Conv2d(512, out_ch, 3, padding=1) self.side6 = nn.Conv2d(512, out_ch, 3, padding=1) self.outconv = nn.Conv2d(6, out_ch, 1) def forward(self, x): x = self.edge(x) hx = x # stage 1 hx1 = self.stage1(hx) hx = self.pool12(hx1) # stage 2 hx2 = self.stage2(hx) hx = self.pool23(hx2) # stage 3 hx3 = self.stage3(hx) hx = self.pool34(hx3) # stage 4 hx4 = self.stage4(hx) hx = self.pool45(hx4) # stage 5 hx5 = self.stage5(hx) hx = self.pool56(hx5) # stage 6 hx6 = self.stage6(hx) hx6up = _upsample_like(hx6, hx5) # -------------------- decoder -------------------- hx5d = self.stage5d(torch.cat((hx6up, hx5), 1)) hx5dup = _upsample_like(hx5d, hx4) hx4d = self.stage4d(torch.cat((hx5dup, hx4), 1)) hx4dup = _upsample_like(hx4d, hx3) hx3d = self.stage3d(torch.cat((hx4dup, hx3), 1)) hx3dup = _upsample_like(hx3d, hx2) hx2d = self.stage2d(torch.cat((hx3dup, hx2), 1)) hx2dup = _upsample_like(hx2d, hx1) hx1d = self.stage1d(torch.cat((hx2dup, hx1), 1)) # side output d1 = self.side1(hx1d) d2 = self.side2(hx2d) d2 = _upsample_like(d2, d1) d3 = self.side3(hx3d) d3 = _upsample_like(d3, d1) d4 = self.side4(hx4d) d4 = _upsample_like(d4, d1) d5 = self.side5(hx5d) d5 = _upsample_like(d5, d1) d6 = self.side6(hx6) d6 = _upsample_like(d6, d1) d0 = self.outconv(torch.cat((d1, d2, d3, d4, d5, d6), 1)) return ( torch.sigmoid(d0), torch.sigmoid(d1), torch.sigmoid(d2), torch.sigmoid(d3), torch.sigmoid(d4), torch.sigmoid(d5), torch.sigmoid(d6), ) ### RSU-5 ### class RSU5(nn.Module): # UNet05DRES(nn.Module): def __init__(self, in_ch=3, mid_ch=12, out_ch=3): super(RSU5, self).__init__() self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1) self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1) self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool3 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=1) self.rebnconv5 = REBNCONV(mid_ch, mid_ch, dirate=2) self.rebnconv4d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1) def forward(self, x): hx = x hxin = self.rebnconvin(hx) hx1 = self.rebnconv1(hxin) hx = self.pool1(hx1) hx2 = self.rebnconv2(hx) hx = self.pool2(hx2) hx3 = self.rebnconv3(hx) hx = self.pool3(hx3) hx4 = self.rebnconv4(hx) hx5 = self.rebnconv5(hx4) hx4d = self.rebnconv4d(torch.cat((hx5, hx4), 1)) hx4dup = _upsample_like(hx4d, hx3) hx3d = self.rebnconv3d(torch.cat((hx4dup, hx3), 1)) hx3dup = _upsample_like(hx3d, hx2) hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1)) hx2dup = _upsample_like(hx2d, hx1) hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1)) return hx1d + hxin ### RSU-6 ### class RSU6(nn.Module): # UNet06DRES(nn.Module): def __init__(self, in_ch=3, mid_ch=12, out_ch=3): super(RSU6, self).__init__() self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1) self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1) self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool3 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool4 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv5 = REBNCONV(mid_ch, mid_ch, dirate=1) self.rebnconv6 = REBNCONV(mid_ch, mid_ch, dirate=2) self.rebnconv5d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv4d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1) def forward(self, x): hx = x hxin = self.rebnconvin(hx) hx1 = self.rebnconv1(hxin) hx = self.pool1(hx1) hx2 = self.rebnconv2(hx) hx = self.pool2(hx2) hx3 = self.rebnconv3(hx) hx = self.pool3(hx3) hx4 = self.rebnconv4(hx) hx = self.pool4(hx4) hx5 = self.rebnconv5(hx) hx6 = self.rebnconv6(hx5) hx5d = self.rebnconv5d(torch.cat((hx6, hx5), 1)) hx5dup = _upsample_like(hx5d, hx4) hx4d = self.rebnconv4d(torch.cat((hx5dup, hx4), 1)) hx4dup = _upsample_like(hx4d, hx3) hx3d = self.rebnconv3d(torch.cat((hx4dup, hx3), 1)) hx3dup = _upsample_like(hx3d, hx2) hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1)) hx2dup = _upsample_like(hx2d, hx1) hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1)) return hx1d + hxin ### RSU-7 ### class RSU7(nn.Module): # UNet07DRES(nn.Module): def __init__(self, in_ch=3, mid_ch=12, out_ch=3): super(RSU7, self).__init__() self.rebnconvin = REBNCONV(in_ch, out_ch, dirate=1) self.rebnconv1 = REBNCONV(out_ch, mid_ch, dirate=1) self.pool1 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv2 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool2 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv3 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool3 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv4 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool4 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv5 = REBNCONV(mid_ch, mid_ch, dirate=1) self.pool5 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.rebnconv6 = REBNCONV(mid_ch, mid_ch, dirate=1) self.rebnconv7 = REBNCONV(mid_ch, mid_ch, dirate=2) self.rebnconv6d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv5d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv4d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv3d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv2d = REBNCONV(mid_ch * 2, mid_ch, dirate=1) self.rebnconv1d = REBNCONV(mid_ch * 2, out_ch, dirate=1) def forward(self, x): hx = x hxin = self.rebnconvin(hx) hx1 = self.rebnconv1(hxin) hx = self.pool1(hx1) hx2 = self.rebnconv2(hx) hx = self.pool2(hx2) hx3 = self.rebnconv3(hx) hx = self.pool3(hx3) hx4 = self.rebnconv4(hx) hx = self.pool4(hx4) hx5 = self.rebnconv5(hx) hx = self.pool5(hx5) hx6 = self.rebnconv6(hx) hx7 = self.rebnconv7(hx6) hx6d = self.rebnconv6d(torch.cat((hx7, hx6), 1)) hx6dup = _upsample_like(hx6d, hx5) hx5d = self.rebnconv5d(torch.cat((hx6dup, hx5), 1)) hx5dup = _upsample_like(hx5d, hx4) hx4d = self.rebnconv4d(torch.cat((hx5dup, hx4), 1)) hx4dup = _upsample_like(hx4d, hx3) hx3d = self.rebnconv3d(torch.cat((hx4dup, hx3), 1)) hx3dup = _upsample_like(hx3d, hx2) hx2d = self.rebnconv2d(torch.cat((hx3dup, hx2), 1)) hx2dup = _upsample_like(hx2d, hx1) hx1d = self.rebnconv1d(torch.cat((hx2dup, hx1), 1)) return hx1d + hxin class U2NETP(nn.Module): def __init__(self, in_ch=3, out_ch=1): super(U2NETP, self).__init__() self.stage1 = RSU7(in_ch, 16, 64) self.pool12 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage2 = RSU6(64, 16, 64) self.pool23 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage3 = RSU5(64, 16, 64) self.pool34 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage4 = RSU4(64, 16, 64) self.pool45 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage5 = RSU4F(64, 16, 64) self.pool56 = nn.MaxPool2d(2, stride=2, ceil_mode=True) self.stage6 = RSU4F(64, 16, 64) # decoder self.stage5d = RSU4F(128, 16, 64) self.stage4d = RSU4(128, 16, 64) self.stage3d = RSU5(128, 16, 64) self.stage2d = RSU6(128, 16, 64) self.stage1d = RSU7(128, 16, 64) self.side1 = nn.Conv2d(64, out_ch, 3, padding=1) self.side2 = nn.Conv2d(64, out_ch, 3, padding=1) self.side3 = nn.Conv2d(64, out_ch, 3, padding=1) self.side4 = nn.Conv2d(64, out_ch, 3, padding=1) self.side5 = nn.Conv2d(64, out_ch, 3, padding=1) self.side6 = nn.Conv2d(64, out_ch, 3, padding=1) self.outconv = nn.Conv2d(6, out_ch, 1) def forward(self, x): hx = x # stage 1 hx1 = self.stage1(hx) hx = self.pool12(hx1) # stage 2 hx2 = self.stage2(hx) hx = self.pool23(hx2) # stage 3 hx3 = self.stage3(hx) hx = self.pool34(hx3) # stage 4 hx4 = self.stage4(hx) hx = self.pool45(hx4) # stage 5 hx5 = self.stage5(hx) hx = self.pool56(hx5) # stage 6 hx6 = self.stage6(hx) hx6up = _upsample_like(hx6, hx5) # decoder hx5d = self.stage5d(torch.cat((hx6up, hx5), 1)) hx5dup = _upsample_like(hx5d, hx4) hx4d = self.stage4d(torch.cat((hx5dup, hx4), 1)) hx4dup = _upsample_like(hx4d, hx3) hx3d = self.stage3d(torch.cat((hx4dup, hx3), 1)) hx3dup = _upsample_like(hx3d, hx2) hx2d = self.stage2d(torch.cat((hx3dup, hx2), 1)) hx2dup = _upsample_like(hx2d, hx1) hx1d = self.stage1d(torch.cat((hx2dup, hx1), 1)) # side output d1 = self.side1(hx1d) d2 = self.side2(hx2d) d2 = _upsample_like(d2, d1) d3 = self.side3(hx3d) d3 = _upsample_like(d3, d1) d4 = self.side4(hx4d) d4 = _upsample_like(d4, d1) d5 = self.side5(hx5d) d5 = _upsample_like(d5, d1) d6 = self.side6(hx6) d6 = _upsample_like(d6, d1) d0 = self.outconv(torch.cat((d1, d2, d3, d4, d5, d6), 1)) return ( torch.sigmoid(d0), torch.sigmoid(d1), torch.sigmoid(d2), torch.sigmoid(d3), torch.sigmoid(d4), torch.sigmoid(d5), torch.sigmoid(d6), )