|
| 1 | +from functools import partial |
| 2 | + |
| 3 | +import torch |
| 4 | +from torch import nn, einsum |
| 5 | + |
| 6 | +from einops import rearrange, repeat |
| 7 | +from einops.layers.torch import Rearrange, Reduce |
| 8 | + |
| 9 | +# helpers |
| 10 | + |
| 11 | +def exists(val): |
| 12 | + return val is not None |
| 13 | + |
| 14 | +def default(val, d): |
| 15 | + return val if exists(val) else d |
| 16 | + |
| 17 | +def cast_tuple(val, length = 1): |
| 18 | + return val if isinstance(val, tuple) else ((val,) * length) |
| 19 | + |
| 20 | +# helper classes |
| 21 | + |
| 22 | +class PreNormResidual(nn.Module): |
| 23 | + def __init__(self, dim, fn): |
| 24 | + super().__init__() |
| 25 | + self.norm = nn.LayerNorm(dim) |
| 26 | + self.fn = fn |
| 27 | + |
| 28 | + def forward(self, x): |
| 29 | + return self.fn(self.norm(x)) + x |
| 30 | + |
| 31 | +# MBConv |
| 32 | + |
| 33 | +class SqueezeExcitation(nn.Module): |
| 34 | + def __init__(self, dim, shrinkage_rate = 0.25): |
| 35 | + super().__init__() |
| 36 | + hidden_dim = int(dim * shrinkage_rate) |
| 37 | + |
| 38 | + self.gate = nn.Sequential( |
| 39 | + Reduce('b c h w -> b c', 'mean'), |
| 40 | + nn.Linear(dim, hidden_dim, bias = False), |
| 41 | + nn.SiLU(), |
| 42 | + nn.Linear(hidden_dim, dim, bias = False), |
| 43 | + nn.Sigmoid(), |
| 44 | + Rearrange('b c -> b c 1 1') |
| 45 | + ) |
| 46 | + |
| 47 | + def forward(self, x): |
| 48 | + return x * self.gate(x) |
| 49 | + |
| 50 | + |
| 51 | +class MBConvResidual(nn.Module): |
| 52 | + def __init__(self, fn, dropout = 0.): |
| 53 | + super().__init__() |
| 54 | + self.fn = fn |
| 55 | + self.dropsample = Dropsample(dropout) |
| 56 | + |
| 57 | + def forward(self, x): |
| 58 | + out = self.fn(x) |
| 59 | + out = self.dropsample(out) |
| 60 | + return out |
| 61 | + |
| 62 | +class Dropsample(nn.Module): |
| 63 | + def __init__(self, prob = 0): |
| 64 | + super().__init__() |
| 65 | + self.prob = prob |
| 66 | + |
| 67 | + def forward(self, x): |
| 68 | + device = x.device |
| 69 | + |
| 70 | + if self.prob == 0. or (not self.training): |
| 71 | + return x |
| 72 | + |
| 73 | + keep_mask = torch.FloatTensor((x.shape[0], 1, 1, 1), device = device).uniform_() > self.prob |
| 74 | + return x * keep_mask / (1 - self.prob) |
| 75 | + |
| 76 | +def MBConv( |
| 77 | + dim_in, |
| 78 | + dim_out, |
| 79 | + *, |
| 80 | + downsample, |
| 81 | + expansion_rate = 4, |
| 82 | + shrinkage_rate = 0.25, |
| 83 | + dropout = 0. |
| 84 | +): |
| 85 | + hidden_dim = int(expansion_rate * dim_out) |
| 86 | + stride = 2 if downsample else 1 |
| 87 | + |
| 88 | + net = nn.Sequential( |
| 89 | + nn.Conv2d(dim_in, dim_out, 1), |
| 90 | + nn.BatchNorm2d(dim_out), |
| 91 | + nn.SiLU(), |
| 92 | + nn.Conv2d(dim_out, dim_out, 3, stride = stride, padding = 1, groups = dim_out), |
| 93 | + SqueezeExcitation(dim_out, shrinkage_rate = shrinkage_rate), |
| 94 | + nn.Conv2d(dim_out, dim_out, 1), |
| 95 | + nn.BatchNorm2d(dim_out) |
| 96 | + ) |
| 97 | + |
| 98 | + if dim_in == dim_out and not downsample: |
| 99 | + net = MBConvResidual(net, dropout = dropout) |
| 100 | + |
| 101 | + return net |
| 102 | + |
| 103 | +# attention related classes |
| 104 | + |
| 105 | +class Attention(nn.Module): |
| 106 | + def __init__( |
| 107 | + self, |
| 108 | + dim, |
| 109 | + dim_head = 32, |
| 110 | + dropout = 0., |
| 111 | + window_size = 7 |
| 112 | + ): |
| 113 | + super().__init__() |
| 114 | + assert (dim % dim_head) == 0, 'dimension should be divisible by dimension per head' |
| 115 | + |
| 116 | + self.heads = dim // dim_head |
| 117 | + self.scale = dim_head ** -0.5 |
| 118 | + |
| 119 | + self.to_qkv = nn.Linear(dim, dim * 3, bias = False) |
| 120 | + |
| 121 | + self.attend = nn.Sequential( |
| 122 | + nn.Softmax(dim = -1), |
| 123 | + nn.Dropout(dropout) |
| 124 | + ) |
| 125 | + |
| 126 | + self.to_out = nn.Sequential( |
| 127 | + nn.Linear(dim, dim, bias = False), |
| 128 | + nn.Dropout(dropout) |
| 129 | + ) |
| 130 | + |
| 131 | + # relative positional bias |
| 132 | + |
| 133 | + self.rel_pos_bias = nn.Embedding((2 * window_size - 1) ** 2, self.heads) |
| 134 | + |
| 135 | + pos = torch.arange(window_size) |
| 136 | + grid = torch.stack(torch.meshgrid(pos, pos, indexing = 'ij')) |
| 137 | + grid = rearrange(grid, 'c i j -> (i j) c') |
| 138 | + rel_pos = rearrange(grid, 'i ... -> i 1 ...') - rearrange(grid, 'j ... -> 1 j ...') |
| 139 | + rel_pos += window_size - 1 |
| 140 | + rel_pos_indices = (rel_pos * torch.tensor([2 * window_size - 1, 1])).sum(dim = -1) |
| 141 | + |
| 142 | + self.register_buffer('rel_pos_indices', rel_pos_indices, persistent = False) |
| 143 | + |
| 144 | + def forward(self, x): |
| 145 | + batch, height, width, window_height, window_width, _, device, h = *x.shape, x.device, self.heads |
| 146 | + |
| 147 | + # flatten |
| 148 | + |
| 149 | + x = rearrange(x, 'b x y w1 w2 d -> (b x y) (w1 w2) d') |
| 150 | + |
| 151 | + # project for queries, keys, values |
| 152 | + |
| 153 | + q, k, v = self.to_qkv(x).chunk(3, dim = -1) |
| 154 | + |
| 155 | + # split heads |
| 156 | + |
| 157 | + q, k, v = map(lambda t: rearrange(t, 'b n (h d ) -> b h n d', h = h), (q, k, v)) |
| 158 | + |
| 159 | + # scale |
| 160 | + |
| 161 | + q = q * self.scale |
| 162 | + |
| 163 | + # sim |
| 164 | + |
| 165 | + sim = einsum('b h i d, b h j d -> b h i j', q, k) |
| 166 | + |
| 167 | + # add positional bias |
| 168 | + |
| 169 | + bias = self.rel_pos_bias(self.rel_pos_indices) |
| 170 | + sim = sim + rearrange(bias, 'i j h -> h i j') |
| 171 | + |
| 172 | + # attention |
| 173 | + |
| 174 | + attn = self.attend(sim) |
| 175 | + |
| 176 | + # aggregate |
| 177 | + |
| 178 | + out = einsum('b h i j, b h j d -> b h i d', attn, v) |
| 179 | + |
| 180 | + # merge heads |
| 181 | + |
| 182 | + out = rearrange(out, 'b h (w1 w2) d -> b w1 w2 (h d)', w1 = window_height, w2 = window_width) |
| 183 | + |
| 184 | + # combine heads out |
| 185 | + |
| 186 | + out = self.to_out(out) |
| 187 | + return rearrange(out, '(b x y) ... -> b x y ...', x = height, y = width) |
| 188 | + |
| 189 | +class MaxViT(nn.Module): |
| 190 | + def __init__( |
| 191 | + self, |
| 192 | + *, |
| 193 | + num_classes, |
| 194 | + dim, |
| 195 | + depth, |
| 196 | + dim_head = 32, |
| 197 | + dim_conv_stem = None, |
| 198 | + window_size = 7, |
| 199 | + mbconv_expansion_rate = 4, |
| 200 | + mbconv_shrinkage_rate = 0.25, |
| 201 | + dropout = 0.1, |
| 202 | + channels = 3 |
| 203 | + ): |
| 204 | + super().__init__() |
| 205 | + assert isinstance(depth, tuple), 'depth needs to be tuple if integers indicating number of transformer blocks at that stage' |
| 206 | + |
| 207 | + # convolutional stem |
| 208 | + |
| 209 | + dim_conv_stem = default(dim_conv_stem, dim) |
| 210 | + |
| 211 | + self.conv_stem = nn.Sequential( |
| 212 | + nn.Conv2d(channels, dim_conv_stem, 3, stride = 2, padding = 1), |
| 213 | + nn.Conv2d(dim_conv_stem, dim_conv_stem, 3, padding = 1) |
| 214 | + ) |
| 215 | + |
| 216 | + # variables |
| 217 | + |
| 218 | + num_stages = len(depth) |
| 219 | + |
| 220 | + dims = tuple(map(lambda i: (2 ** i) * dim, range(num_stages))) |
| 221 | + dims = (dim_conv_stem, *dims) |
| 222 | + dim_pairs = tuple(zip(dims[:-1], dims[1:])) |
| 223 | + |
| 224 | + self.layers = nn.ModuleList([]) |
| 225 | + |
| 226 | + # shorthand for window size for efficient block - grid like attention |
| 227 | + |
| 228 | + w = window_size |
| 229 | + |
| 230 | + # iterate through stages |
| 231 | + |
| 232 | + for ind, ((layer_dim_in, layer_dim), layer_depth) in enumerate(zip(dim_pairs, depth)): |
| 233 | + for stage_ind in range(layer_depth): |
| 234 | + is_first = stage_ind == 0 |
| 235 | + stage_dim_in = layer_dim_in if is_first else layer_dim |
| 236 | + |
| 237 | + block = nn.Sequential( |
| 238 | + MBConv( |
| 239 | + stage_dim_in, |
| 240 | + layer_dim, |
| 241 | + downsample = is_first, |
| 242 | + expansion_rate = mbconv_expansion_rate, |
| 243 | + shrinkage_rate = mbconv_shrinkage_rate |
| 244 | + ), |
| 245 | + Rearrange('b d (x w1) (y w2) -> b x y w1 w2 d', w1 = w, w2 = w), # block-like attention |
| 246 | + PreNormResidual(layer_dim, Attention(dim = layer_dim, dim_head = dim_head, dropout = dropout, window_size = w)), |
| 247 | + Rearrange('b x y w1 w2 d -> b d (x w1) (y w2)'), |
| 248 | + |
| 249 | + Rearrange('b d (w1 x) (w2 y) -> b x y w1 w2 d', w1 = w, w2 = w), # grid-like attention |
| 250 | + PreNormResidual(layer_dim, Attention(dim = layer_dim, dim_head = dim_head, dropout = dropout, window_size = w)), |
| 251 | + Rearrange('b x y w1 w2 d -> b d (w1 x) (w2 y)'), |
| 252 | + ) |
| 253 | + |
| 254 | + self.layers.append(block) |
| 255 | + |
| 256 | + # mlp head out |
| 257 | + |
| 258 | + self.mlp_head = nn.Sequential( |
| 259 | + Reduce('b d h w -> b d', 'mean'), |
| 260 | + nn.LayerNorm(dims[-1]), |
| 261 | + nn.Linear(dims[-1], num_classes) |
| 262 | + ) |
| 263 | + |
| 264 | + def forward(self, x): |
| 265 | + x = self.conv_stem(x) |
| 266 | + |
| 267 | + for stage in self.layers: |
| 268 | + x = stage(x) |
| 269 | + |
| 270 | + return self.mlp_head(x) |
0 commit comments