TTXian/RemoteSensingChangeDetection-RSCD.HA2F
0
1# Copyright (c) Meta Platforms, Inc. and affiliates.2#3# This source code is licensed under the Apache License, Version 2.04# found in the LICENSE file in the root directory of this source tree.5 6# References:7# https://github.com/facebookresearch/dino/blob/main/vision_transformer.py8# https://github.com/rwightman/pytorch-image-models/tree/master/timm/models/vision_transformer.py9 10from functools import partial11import math12import logging13from typing import Sequence, Tuple, Union, Callable14 15import torch16import torch.nn as nn17import torch.utils.checkpoint18from torch.nn.init import trunc_normal_19from einops import rearrange20 21from model.layers import Mlp, PatchEmbed, SwiGLUFFNFused, MemEffAttention, NestedTensorBlock as Block22from model.resnet import resnet1823 24 25def named_apply(fn: Callable, module: nn.Module, name="", depth_first=True, include_root=False) -> nn.Module:26 if not depth_first and include_root:27 fn(module=module, name=name)28 for child_name, child_module in module.named_children():29 child_name = ".".join((name, child_name)) if name else child_name30 named_apply(fn=fn, module=child_module, name=child_name, depth_first=depth_first, include_root=True)31 if depth_first and include_root:32 fn(module=module, name=name)33 return module34 35 36class BlockChunk(nn.ModuleList):37 def forward(self, x):38 for b in self:39 x = b(x)40 return x41 42 43class DinoVisionTransformer(nn.Module):44 def __init__(45 self,46 img_size=224,47 patch_size=16,48 in_chans=3,49 embed_dim=768,50 depth=12,51 num_heads=12,52 mlp_ratio=4.0,53 qkv_bias=True,54 ffn_bias=True,55 proj_bias=True,56 drop_path_rate=0.0,57 drop_path_uniform=False,58 init_values=None, # for layerscale: None or 0 => no layerscale59 embed_layer=PatchEmbed,60 act_layer=nn.GELU,61 block_fn=Block,62 ffn_layer="mlp",63 block_chunks=0,64 num_register_tokens=0,65 interpolate_antialias=False,66 interpolate_offset=0.1,67 ):68 """69 Args:70 img_size (int, tuple): input image size71 patch_size (int, tuple): patch size72 in_chans (int): number of input channels73 embed_dim (int): embedding dimension74 depth (int): depth of transformer75 num_heads (int): number of attention heads76 mlp_ratio (int): ratio of mlp hidden dim to embedding dim77 qkv_bias (bool): enable bias for qkv if True78 proj_bias (bool): enable bias for proj in attn if True79 ffn_bias (bool): enable bias for ffn if True80 drop_path_rate (float): stochastic depth rate81 drop_path_uniform (bool): apply uniform drop rate across blocks82 weight_init (str): weight init scheme83 init_values (float): layer-scale init values84 embed_layer (nn.Module): patch embedding layer85 act_layer (nn.Module): MLP activation layer86 block_fn (nn.Module): transformer block class87 ffn_layer (str): "mlp", "swiglu", "swiglufused" or "identity"88 block_chunks: (int) split block sequence into block_chunks units for FSDP wrap89 num_register_tokens: (int) number of extra cls tokens (so-called "registers")90 interpolate_antialias: (str) flag to apply anti-aliasing when interpolating positional embeddings91 interpolate_offset: (float) work-around offset to apply when interpolating positional embeddings92 """93 super().__init__()94 norm_layer = partial(nn.LayerNorm, eps=1e-6)95 96 self.num_features = self.embed_dim = embed_dim # num_features for consistency with other models97 self.n_blocks = depth98 self.num_heads = num_heads99 self.patch_size = patch_size100 self.num_register_tokens = num_register_tokens101 self.interpolate_antialias = interpolate_antialias102 self.interpolate_offset = interpolate_offset103 104 self.patch_embed = embed_layer(img_size=img_size, patch_size=patch_size, in_chans=in_chans, embed_dim=embed_dim)105 num_patches = self.patch_embed.num_patches106 107 self.pos_embed = nn.Parameter(torch.zeros(1, num_patches, embed_dim))108 assert num_register_tokens >= 0109 self.register_tokens = (110 nn.Parameter(torch.zeros(1, num_register_tokens, embed_dim)) if num_register_tokens else None111 )112 113 if drop_path_uniform is True:114 dpr = [drop_path_rate] * depth115 else:116 dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule117 118 if ffn_layer == "mlp":119 print("using MLP layer as FFN")120 ffn_layer = Mlp121 elif ffn_layer == "swiglufused" or ffn_layer == "swiglu":122 print("using SwiGLU layer as FFN")123 ffn_layer = SwiGLUFFNFused124 elif ffn_layer == "identity":125 print("using Identity layer as FFN")126 127 def f(*args, **kwargs):128 return nn.Identity()129 130 ffn_layer = f131 else:132 raise NotImplementedError133 134 blocks_list = [135 block_fn(136 dim=embed_dim,137 num_heads=num_heads,138 mlp_ratio=mlp_ratio,139 qkv_bias=qkv_bias,140 proj_bias=proj_bias,141 ffn_bias=ffn_bias,142 drop_path=dpr[i],143 norm_layer=norm_layer,144 act_layer=act_layer,145 ffn_layer=ffn_layer,146 init_values=init_values,147 )148 for i in range(depth)149 ]150 if block_chunks > 0:151 self.chunked_blocks = True152 chunked_blocks = []153 chunksize = depth // block_chunks154 for i in range(0, depth, chunksize):155 # this is to keep the block index consistent if we chunk the block list156 chunked_blocks.append([nn.Identity()] * i + blocks_list[i: i + chunksize])157 self.blocks = nn.ModuleList([BlockChunk(p) for p in chunked_blocks])158 else:159 self.chunked_blocks = False160 self.blocks = nn.ModuleList(blocks_list)161 162 self.norm = norm_layer(embed_dim)163 self.head = nn.Identity()164 165 self.mask_token = nn.Parameter(torch.zeros(1, embed_dim))166 167 self.init_weights()168 169 def init_weights(self):170 trunc_normal_(self.pos_embed, std=0.02)171 if self.register_tokens is not None:172 nn.init.normal_(self.register_tokens, std=1e-6)173 named_apply(init_weights_vit_timm, self)174 175 def interpolate_pos_encoding(self, x, w, h):176 previous_dtype = x.dtype177 npatch = x.shape[1] - 1178 N = self.pos_embed.shape[1]179 if npatch == N and w == h:180 return self.pos_embed181 patch_pos_embed = self.pos_embed.float()182 dim = x.shape[-1]183 w0 = w // self.patch_size184 h0 = h // self.patch_size185 # we add a small number to avoid floating point error in the interpolation186 # see discussion at https://github.com/facebookresearch/dino/issues/8187 w0, h0 = w0 + self.interpolate_offset, h0 + self.interpolate_offset188 189 sqrt_N = math.sqrt(N)190 sx, sy = float(w0) / sqrt_N, float(h0) / sqrt_N191 patch_pos_embed = nn.functional.interpolate(192 patch_pos_embed.reshape(1, int(sqrt_N), int(sqrt_N), dim).permute(0, 3, 1, 2),193 scale_factor=(sx, sy),194 mode="bicubic",195 antialias=self.interpolate_antialias,196 )197 198 assert int(w0) == patch_pos_embed.shape[-2]199 assert int(h0) == patch_pos_embed.shape[-1]200 patch_pos_embed = patch_pos_embed.permute(0, 2, 3, 1).view(1, -1, dim)201 return patch_pos_embed.to(previous_dtype)202 203 def prepare_tokens_with_masks(self, x, masks=None):204 B, nc, w, h = x.shape205 x = self.patch_embed(x)206 if masks is not None:207 x = torch.where(masks.unsqueeze(-1), self.mask_token.to(x.dtype).unsqueeze(0), x)208 209 x = x + self.interpolate_pos_encoding(x, w, h)210 211 if self.register_tokens is not None:212 x = torch.cat(213 (214 x[:, :1],215 self.register_tokens.expand(x.shape[0], -1, -1),216 x[:, 1:],217 ),218 dim=1,219 )220 221 return x222 223 def forward_features_list(self, x_list, masks_list):224 x = [self.prepare_tokens_with_masks(x, masks) for x, masks in zip(x_list, masks_list)]225 for blk in self.blocks:226 x = blk(x)227 228 all_x = x229 output = []230 for x, masks in zip(all_x, masks_list):231 x_norm = self.norm(x)232 output.append(233 {234 "x_norm_clstoken": x_norm[:, 0],235 "x_norm_regtokens": x_norm[:, 1: self.num_register_tokens + 1],236 "x_norm_patchtokens": x_norm[:, self.num_register_tokens + 1:],237 "x_prenorm": x,238 "masks": masks,239 }240 )241 return output242 243 def forward(self, x, masks=None):244 if isinstance(x, list):245 return self.forward_features_list(x, masks)246 247 x = self.prepare_tokens_with_masks(x, masks)248 249 for blk in self.blocks:250 x = blk(x)251 252 x_norm = self.norm(x)253 return x_norm254 255 def _get_intermediate_layers_not_chunked(self, x, n=1):256 x = self.prepare_tokens_with_masks(x)257 # If n is an int, take the n last blocks. If it's a list, take them258 output, total_block_len = [], len(self.blocks)259 blocks_to_take = range(total_block_len - n, total_block_len) if isinstance(n, int) else n260 for i, blk in enumerate(self.blocks):261 x = blk(x)262 if i in blocks_to_take:263 output.append(x)264 assert len(output) == len(blocks_to_take), f"only {len(output)} / {len(blocks_to_take)} blocks found"265 return output266 267 def _get_intermediate_layers_chunked(self, x, n=1):268 x = self.prepare_tokens_with_masks(x)269 output, i, total_block_len = [], 0, len(self.blocks[-1])270 # If n is an int, take the n last blocks. If it's a list, take them271 blocks_to_take = range(total_block_len - n, total_block_len) if isinstance(n, int) else n272 for block_chunk in self.blocks:273 for blk in block_chunk[i:]: # Passing the nn.Identity()274 x = blk(x)275 if i in blocks_to_take:276 output.append(x)277 i += 1278 assert len(output) == len(blocks_to_take), f"only {len(output)} / {len(blocks_to_take)} blocks found"279 return output280 281 def get_intermediate_layers(282 self,283 x: torch.Tensor,284 n: Union[int, Sequence] = 1, # Layers or n last layers to take285 reshape: bool = False,286 return_class_token: bool = False,287 norm=True,288 ) -> Tuple[Union[torch.Tensor, Tuple[torch.Tensor]]]:289 if self.chunked_blocks:290 outputs = self._get_intermediate_layers_chunked(x, n)291 else:292 outputs = self._get_intermediate_layers_not_chunked(x, n)293 if norm:294 outputs = [self.norm(out) for out in outputs]295 class_tokens = [out[:, 0] for out in outputs]296 outputs = [out[:, 1 + self.num_register_tokens:] for out in outputs]297 if reshape:298 B, _, w, h = x.shape299 outputs = [300 out.reshape(B, w // self.patch_size, h // self.patch_size, -1).permute(0, 3, 1, 2).contiguous()301 for out in outputs302 ]303 if return_class_token:304 return tuple(zip(outputs, class_tokens))305 return tuple(outputs)306 307 308def init_weights_vit_timm(module: nn.Module, name: str = ""):309 """ViT weight initialization, original timm impl (for reproducibility)"""310 if isinstance(module, nn.Linear):311 trunc_normal_(module.weight, std=0.02)312 if module.bias is not None:313 nn.init.zeros_(module.bias)314 315 316class Encoder(nn.Module):317 def __init__(self, model_type='small'):318 super().__init__()319 if model_type == 'tiny':320 self.vit = DinoVisionTransformer(321 img_size=256,322 patch_size=16,323 embed_dim=192,324 depth=12,325 num_heads=6,326 mlp_ratio=4,327 block_fn=partial(Block, attn_class=MemEffAttention),328 num_register_tokens=0329 )330 path = "checkpoint/deit_tiny_patch16_224-a1311bcf.pth"331 332 elif model_type == 'small':333 self.vit = DinoVisionTransformer(334 img_size=256,335 patch_size=16,336 embed_dim=384,337 depth=12,338 num_heads=6,339 mlp_ratio=4,340 block_fn=partial(Block, attn_class=MemEffAttention),341 num_register_tokens=0342 )343 path = "checkpoint/dinov2_vits14_pretrain.pth"344 345 else:346 assert False, r'Encoder: check the vit model type'347 348 state_dict = torch.load(path, map_location='cpu')['model'] \349 if model_type == 'tiny' else torch.load(path, map_location='cpu')350 351 for k in ['pos_embed', 'patch_embed.proj.weight']:352 del state_dict[k]353 msg = self.vit.load_state_dict(state_dict, strict=False)354 print(' missing_keys:{},\n unexpected_keys:{}'.format(msg.missing_keys, msg.unexpected_keys))355 print('model_type: {},\n checkpoint_path: {}'.format(model_type, path))356 357 self.resnet = resnet18(pretrained=True)358 self.drop = nn.Dropout(p=0.01)359 360 # 新增特征融合模块361 self.fusion_conv = nn.Sequential(362 nn.Conv2d(512 + 384, 384, kernel_size=1), # 假设ViT embed_dim=384363 nn.BatchNorm2d(384),364 nn.ReLU(inplace=True)365 )366 367 def detail_capture(self, x):368 x = self.resnet.conv1(x)369 x = self.resnet.bn1(x)370 x = self.resnet.relu(x)371 372 x2 = self.drop(self.resnet.layer1(x))373 x3 = self.resnet.layer2(x2)374 x4 = self.resnet.layer3(x3)375 x5 = self.resnet.layer4(x4)376 return [x2, x3, x4, x5]377 378 def forward(self, x, y):379 380 v_x = self.vit(x)381 v_y = self.vit(y)382 383 v_x = rearrange(v_x, 'b (h w) c -> b c h w', h=16, w=16)384 v_y = rearrange(v_y, 'b (h w) c -> b c h w', h=16, w=16)385 386 c_x = self.detail_capture(x)387 c_y = self.detail_capture(y)388 389 fused_v_x = self.fusion_conv(torch.cat([c_x[-1], v_x], dim=1))390 fused_v_y = self.fusion_conv(torch.cat([c_y[-1], v_y], dim=1))391 return c_x[:-1] + [fused_v_x], c_y[:-1] + [fused_v_y]392