RuiTerrty/RemoteSensingChangeDetection-RSCD.HA2F

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