参考"“”
Basic mamda_fetrack model.
“”“
import math
import os
from typing import List

import torch
from torch import nn
from torch.nn.modules.transformer import _get_clones

from lib.models.layers.head import build_box_head
from lib.utils.box_ops import box_xyxy_to_cxcywh
from lib.models.mamba_fetrack.models_mamba import create_block
from timm.models import create_model
import torch.nn.functional as F
from lib.models.mamba_fetrack.mamba_cross import CrossMamba
from thop import profile


class Mamba_FEtrack(nn.Module):
“”” This is the base class for mamda_fetrack “”“

def init(self, visionmamba, cross_mamba, box_head, aux_loss=False, head_type=“CORNER”):
“”” Initializes the model.
Parameters:
transformer: torch module of the transformer architecture.
aux_loss: True if auxiliary decoding losses (loss at each decoder layer) are to be used.
“”“
super().init()
self.backbone = visionmamba
self.cross_mamba= cross_mamba
self.box_head = box_head

self.aux_loss = aux_loss
self.head_type = head_type
if head_type == “CORNER” or head_type == “CENTER”:
self.feat_sz_s = int(box_head.feat_sz)
self.feat_len_s = int(box_head.feat_sz ** 2)

if self.aux_loss:
self.box_head = _get_clones(self.box_head, 6)

def forward(self, template: torch.Tensor,
search: torch.Tensor,
event_template: torch.Tensor,
event_search: torch.Tensor,
ce_template_mask=None,
ce_keep_rate=None,
return_last_attn=False,
):

rgb_feature = self.backbone.forward_features( z=template, x=search, #[B, 320, 384]
inference_params=None, if_random_cls_token_position=False, if_random_token_rank=False)
event_feature = self.backbone.forward_features(z=event_template, x=event_search, #[B, 320, 384]
inference_params=None, if_random_cls_token_position=False, if_random_token_rank=False)

residual_event_f = 0
residual_rgb_f = 0
event_f = self.cross_mamba(event_feature,residual_event_f,rgb_feature) + event_feature
rgb_f = self.cross_mamba(rgb_feature,residual_rgb_f,event_feature) + rgb_feature

event_searh = event_f[:, -self.feat_len_s:]
rgb_search = rgb_f[:, -self.feat_len_s:]
x = torch.cat((event_searh,rgb_search),dim=-1)


# Forward head
feat_last = x
if isinstance(x, list):
feat_last = x[-1]
out = self.forward_head(feat_last, None)

out[‘backbone_feat’] = x
return out

def forward_head(self, cat_feature, gt_score_map=None):
“””
cat_feature: output embeddings of the backbone, it can be (HW1+HW2, B, C) or (HW2, B, C)
“”“
search_feature = cat_feature
opt = (search_feature.unsqueeze(-1)).permute((0, 3, 2, 1)).contiguous() # opt.shape = torch.Size([B, 1, 384, 256])
bs, Nq, C, HW = opt.size()
opt_feat = opt.view(-1, C, self.feat_sz_s, self.feat_sz_s) # opt_feat.shape = torch.Size([B, 384, 16, 16])

if self.head_type == “CORNER”:
# run the corner head
pred_box, score_map = self.box_head(opt_feat, True)
outputs_coord = box_xyxy_to_cxcywh(pred_box)
outputs_coord_new = outputs_coord.view(bs, Nq, 4)
out = {‘pred_boxes’: outputs_coord_new,
‘score_map’: score_map,
}
return out

elif self.head_type == “CENTER”:
# run the center head

score_map_ctr, bbox, size_map, offset_map = self.box_head(opt_feat, gt_score_map)
outputs_coord = bbox
outputs_coord_new = outputs_coord.view(bs, Nq, 4)
out = {‘pred_boxes’: outputs_coord_new,
‘score_map’: score_map_ctr,
‘size_map’: size_map,
‘offset_map’: offset_map}
return out
else:
raise NotImplementedError


def build_mamba_fetrack(cfg, training=True):
current_dir = os.path.dirname(os.path.abspath(file)) # This is your Project Root
pretrained_path = os.path.join(current_dir, ‘…/…/…/pretrained_models’)
if cfg.MODEL.PRETRAIN_FILE and (‘Mamba_FETrack’ not in cfg.MODEL.PRETRAIN_FILE) and training:
pretrained = os.path.join(pretrained_path, cfg.MODEL.PRETRAIN_FILE)
else:
pretrained = ‘’

backbone = create_model( model_name= cfg.MODEL.BACKBONE.TYPE, pretrained= pretrained, num_classes=1000,
drop_rate=0.0, drop_path_rate=cfg.TRAIN.DROP_PATH_RATE, drop_block_rate=None, img_size=256
)
hidden_dim = 384
cross_mamba = CrossMamba(hidden_dim)
box_head = build_box_head(cfg, hidden_dim2)
model = Mamba_FEtrack(
backbone,
cross_mamba,
box_head,
aux_loss=False,
head_type=cfg.MODEL.HEAD.TYPE,
)

if ‘Mamba_FETrack’ in cfg.MODEL.PRETRAIN_FILE and training:
checkpoint = torch.load(cfg.MODEL.PRETRAIN_FILE, map_location=“cpu”)
missing_keys, unexpected_keys = model.load_state_dict(checkpoint[“net”], strict=False)
print('Load pretrained model from: ’ + cfg.MODEL.PRETRAIN_FILE)

return model

，其中 import math
import torch
import torch.nn as nn
import torch.nn.functional as F
from einops import rearrange
import numbers
from .mamba_cross_simple import Mamba
# 将输入张量从（b,c,h,w)重排为（b,hw,c)，常用于将图像张量转换为序列格式
def to_3d(x):
return rearrange(x, ‘b c h w -> b (h w) c’)
# 将序列张量从（b,h*w,c)重排回（b,c,h,w)，这是to_3d的反向操作,通常在完成序列处理之后用于恢复原始形状。
def to_4d(x, h, w):
return rearrange(x, ‘b (h w) c -> b c h w’, h=h, w=w)
# 无偏LayerNorm
# 实现了不包含偏置项的层标准化,计算输入的方差，用于对输入进行规范化处理，再通过权重进行缩放。
class BiasFree_LayerNorm(nn.Module):
def init(self, normalized_shape):
super(BiasFree_LayerNorm, self).init()
if isinstance(normalized_shape, numbers.Integral):
normalized_shape = (normalized_shape,)
normalized_shape = torch.Size(normalized_shape)

assert len(normalized_shape) == 1

self.weight = nn.Parameter(torch.ones(normalized_shape))
self.normalized_shape = normalized_shape

def forward(self, x):
sigma = x.var(-1, keepdim=True, unbiased=False)
return x / torch.sqrt(sigma+1e-5) * self.weight

# 包含偏置的层标准化
class WithBias_LayerNorm(nn.Module):
def init(self, normalized_shape):
super(WithBias_LayerNorm, self).init()
if isinstance(normalized_shape, numbers.Integral):
normalized_shape = (normalized_shape,)
normalized_shape = torch.Size(normalized_shape)

assert len(normalized_shape) == 1

self.weight = nn.Parameter(torch.ones(normalized_shape))
self.bias = nn.Parameter(torch.zeros(normalized_shape))
self.normalized_shape = normalized_shape

def forward(self, x):
mu = x.mean(-1, keepdim=True)
sigma = x.var(-1, keepdim=True, unbiased=False)
return (x - mu) / torch.sqrt(sigma+1e-5) * self.weight + self.bias

# 对2种不同的层标准化进行的封装 LayerNorm_type参数决定了使用哪种层标准化
# 同时通过先把二维图像数据转换为3维序列，应用层标准化再转换回原来的2维结构。
class LayerNorm(nn.Module):
def init(self, dim, LayerNorm_type):
super(LayerNorm, self).init()
if LayerNorm_type ==‘BiasFree’:
self.body = BiasFree_LayerNorm(dim)
else:
self.body = WithBias_LayerNorm(dim)

def forward(self, x):
h, w = x.shape[-2:]
return to_4d(self.body(to_3d(x)), h, w)

# CrossMamba是Mamba的变体，bimamba_type='V3’指明了具体的Mmaba版本
# self.cross_mamba()用于处理来自两个不同模态的特征，m1和m2分别是两个模态的特征, m1_resi是m1的残差,并输出融合后的全局特征global_f
# self.norm1和self.norm2分别对m1和m2进行层标准化处理（包含偏置）
# self.dwconv是一个深度可分离卷积层,用于局部空间内增强特征表达能力。
# 先 to_3d 再 to_4d 操作,避免了直接在多维空间上操作的复杂性,并确保了层标准化可以适用于不同位置的特征向量，保持处理的一致性和独立性。
class CrossMamba(nn.Module):
def init(self, dim):
super(CrossMamba, self).init()
self.cross_mamba = Mamba(dim,bimamba_type=“v3”) # 使用v3版本,会使用额外的输入extra_emb进行特殊处理
self.norm1 = LayerNorm(dim,‘with_bias’)
self.norm2 = LayerNorm(dim,‘with_bias’)
self.dwconv = nn.Conv2d(dim, dim, kernel_size=3, padding=1, groups=dim)
def forward(self,m1,m1_resi,m2):
# print(‘m1.shape’,m1.shape) # m1.shape torch.Size([4, 320, 384])
# print(‘m2.shape’,m2.shape) # m2.shape torch.Size([4, 320, 384])
# print(m1_resi) # 0
m1_resi = m1+m1_resi
# print(‘m1_resi.shape’,m1_resi.shape) # m2.shape torch.Size([4, 320, 384])

global_f = self.cross_mamba(m1, extra_emb=m2)
# print(‘global_f.shape’,global_f.shape) # global_f.shape torch.Size([4, 320, 384])
return global_f，那么现在在# 将 4输入分开，构建新的相同模态结合的2输入，2分支
import math
import logging
from functools import partial
from collections import OrderedDict
from copy import deepcopy

import torch
import torch.nn as nn
import torch.nn.functional as F

from timm.models.layers import to_2tuple

from lib.models.layers.patch_embed import PatchEmbed, PatchEmbed_event, xcorr_depthwise
from .utils import combine_tokens, recover_tokens
from .vit import VisionTransformer
from …layers.attn_blocks import CEBlock
from .new_counter_guide import Counter_Guide
from .ad_counter_guide import Counter_Guide_Enhanced
from …mamba_fetrack.models_mamba import VisionMamba, vim_small_patch16_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2
# 集成CrossMamba模块进行不同模态特征交互
from …mamba_fetrack.mamba_cross import CrossMamba
_logger = logging.getLogger(name)


class VisionTransformerCE(VisionTransformer):
“”” Vision Transformer with candidate elimination (CE) module

A PyTorch impl of : An Image is Worth 16x16 Words: Transformers for Image Recognition at Scale
- https://arxiv.org/abs/2010.11929

Includes distillation token & head support for DeiT: Data-efficient Image Transformers
- https://arxiv.org/abs/2012.12877
“”“
def init(self, img_size=224, patch_size=16, in_chans=3, num_classes=1000, embed_dim=768, depth=12,
num_heads=12, mlp_ratio=4., qkv_bias=True, representation_size=None, distilled=False,
drop_rate=0., attn_drop_rate=0., drop_path_rate=0., embed_layer=PatchEmbed, norm_layer=None,
act_layer=None, weight_init=‘’,
ce_loc=None, ce_keep_ratio=None):
super().init()
if isinstance(img_size, tuple):
self.img_size = img_size
else:
self.img_size = to_2tuple(img_size)
self.patch_size = patch_size
self.in_chans = in_chans

self.num_classes = num_classes
self.num_features = self.embed_dim = embed_dim # num_features for consistency with other models
self.num_tokens = 2 if distilled else 1
norm_layer = norm_layer or partial(nn.LayerNorm, eps=1e-6)
act_layer = act_layer or nn.GELU

self.patch_embed = embed_layer(
img_size=img_size, patch_size=patch_size, in_chans=in_chans, embed_dim=embed_dim)
num_patches = self.patch_embed.num_patches

self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim))
self.dist_token = nn.Parameter(torch.zeros(1, 1, embed_dim)) if distilled else None
self.pos_embed = nn.Parameter(torch.zeros(1, num_patches + self.num_tokens, embed_dim))
self.pos_drop = nn.Dropout(p=drop_rate)
self.pos_embed_event = PatchEmbed_event(in_chans=32, embed_dim=768, kernel_size=4, stride=4)
dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule
blocks = []
ce_index = 0
self.ce_loc = ce_loc
for i in range(depth):
ce_keep_ratio_i = 1.0
if ce_loc is not None and i in ce_loc:
ce_keep_ratio_i = ce_keep_ratio[ce_index]
ce_index += 1

blocks.append(
CEBlock(
dim=embed_dim, num_heads=num_heads, mlp_ratio=mlp_ratio, qkv_bias=qkv_bias, drop=drop_rate,
attn_drop=attn_drop_rate, drop_path=dpr[i], norm_layer=norm_layer, act_layer=act_layer,
keep_ratio_search=ce_keep_ratio_i)
)

self.blocks = nn.Sequential(*blocks)
self.norm = norm_layer(embed_dim)

self.init_weights(weight_init)

# 添加交互模块counter_guide
self.counter_guide = Counter_Guide_Enhanced(768, 768)
# 初始化CrossMamba模块
self.cross_mamba = CrossMamba(dim=768)
#########################################

def forward_features(self, z, x, event_z, event_x,
mask_z=None, mask_x=None,
ce_template_mask=None, ce_keep_rate=None,
return_last_attn=False
):
m1_resi = torch.zeros_like(x)
# 分支1 处理流程
B, H, W = x.shape[0], x.shape[2], x.shape[3]

x = self.patch_embed(x)
z = self.patch_embed(z)
z += self.pos_embed_z
x += self.pos_embed_x

if mask_z is not None and mask_x is not None:
mask_z = F.interpolate(mask_z[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_z = mask_z.flatten(1).unsqueeze(-1)

mask_x = F.interpolate(mask_x[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_x = mask_x.flatten(1).unsqueeze(-1)

mask_x = combine_tokens(mask_z, mask_x, mode=self.cat_mode)
mask_x = mask_x.squeeze(-1)

if self.add_cls_token:
cls_tokens = self.cls_token.expand(B, -1, -1)
cls_tokens = cls_tokens + self.cls_pos_embed
if self.add_sep_seg:
x += self.search_segment_pos_embed
z += self.template_segment_pos_embed
x = combine_tokens(z, x, mode=self.cat_mode)
if self.add_cls_token:
x = torch.cat([cls_tokens, x], dim=1)
x = self.pos_drop(x)

lens_z = self.pos_embed_z.shape[1]
lens_x = self.pos_embed_x.shape[1]
global_index_t = torch.linspace(0, lens_z - 1, lens_z).to(x.device)
global_index_t = global_index_t.repeat(B, 1)
global_index_s = torch.linspace(0, lens_x - 1, lens_x).to(x.device)
global_index_s = global_index_s.repeat(B, 1)
removed_indexes_s = []
# # 分支2 处理流程
event_x = self.pos_embed_event(event_x)
event_z = self.pos_embed_event(event_z)
event_x += self.pos_embed_x
event_z += self.pos_embed_z

event_x = combine_tokens(event_z, event_x, mode=self.cat_mode)

if self.add_cls_token:
event_x = torch.cat([cls_tokens, event_x], dim=1)

lens_z = self.pos_embed_z.shape[1]
lens_x = self.pos_embed_x.shape[1]

global_index_t1 = torch.linspace(0, lens_z - 1, lens_z).to(event_x.device)
global_index_t1 = global_index_t1.repeat(B, 1)
global_index_s1 = torch.linspace(0, lens_x - 1, lens_x).to(event_x.device)
global_index_s1 = global_index_s1.repeat(B, 1)

removed_indexes_s1 = []

for i, blk in enumerate(self.blocks):
# 第一个分支处理
x, global_index_t, global_index_s, removed_index_s, attn = <br/> blk(x, global_index_t, global_index_s, mask_x, ce_template_mask, ce_keep_rate)
# 第二个分支处理
event_x, global_index_t1, global_index_s1, removed_index_s1, attn1 = <br/> blk(event_x, global_index_t1, global_index_s1, mask_x, ce_template_mask, ce_keep_rate)


if self.ce_loc is not None and i in self.ce_loc:
removed_indexes_s.append(removed_index_s)
removed_indexes_s1.append(removed_index_s1)

# 在第1层增加counter_guide模块,验证早期融合效果
if i == 0 :
# 引入counter_guide进行特征增强
enhanced_x, enhanced_event_x = self.counter_guide(x, event_x)
# 将增强后的特征与原特征相加
x = x + enhanced_x
event_x = event_x + enhanced_event_x
m1_resi = enhanced_x
##############
lens_x_new = global_index_s.shape[1]
lens_z_new = global_index_t.shape[1]
z = x[:, :lens_z_new]
x = x[:, lens_z_new:]
lens_eventx_new = global_index_s1.shape[1]
lens_eventz_new = global_index_t1.shape[1]
event_z = event_x[:, :lens_eventz_new]
event_x = event_x[:, lens_eventz_new:]
## 恢复原始的 token 顺序 #################################
if removed_indexes_s and removed_indexes_s[0] is not None:
removed_indexes_cat = torch.cat(removed_indexes_s, dim=1)
pruned_lens_x = lens_x - lens_x_new
pad_x = torch.zeros([B, pruned_lens_x, x.shape[2]], device=x.device)
x = torch.cat([x, pad_x], dim=1)
index_all = torch.cat([global_index_s, removed_indexes_cat], dim=1)
# 恢复原始顺序
C = x.shape[-1]
x = torch.zeros_like(x).scatter(dim=1, index=index_all.unsqueeze(-1).expand(B, -1, C).to(torch.int64), src=x)
x = recover_tokens(x, lens_z_new, lens_x, mode=self.cat_mode)

if removed_indexes_s1 and removed_indexes_s1[0] is not None:
removed_indexes_cat1 = torch.cat(removed_indexes_s1, dim=1)
pruned_lens_x1 = lens_x - lens_eventx_new
pad_x1 = torch.zeros([B, pruned_lens_x1, event_x.shape[2]], device=event_x.device)
event_x = torch.cat([event_x, pad_x1], dim=1)
index_all1 = torch.cat([global_index_s1, removed_indexes_cat1], dim=1)
# 恢复原始顺序
C1 = event_x.shape[-1]
event_x = torch.zeros_like(event_x).scatter(dim=1, index=index_all1.unsqueeze(-1).expand(B, -1, C1).to(torch.int64), src=event_x)
event_x = recover_tokens(event_x, lens_eventz_new, lens_x, mode=self.cat_mode)
################################################################

## 引入Crossmamba进行事件和图像特征融合,不采用concatenate
x = self.cross_mamba(x, m1_resi, event_x)
# 应用LayerNorm归一化处理
x = self.norm(x)

aux_dict = {
“attn”: attn,
‘attn1’: attn1,
“removed_indexes_s”: removed_indexes_s, # used for visualization
‘removed_indexes_s1’: removed_indexes_s1,
}

return x, aux_dict

def forward(self, z, x, event_z, event_x,
ce_template_mask=None, ce_keep_rate=None,
tnc_keep_rate=None,
return_last_attn=False):

x, aux_dict = self.forward_features(z, x, event_z, event_x, ce_template_mask=ce_template_mask, ce_keep_rate=ce_keep_rate,)

return x, aux_dict


def _create_vision_transformer(pretrained=False, **kwargs):
model = VisionTransformerCE(**kwargs)

if pretrained:
if ‘npz’ in pretrained:
model.load_pretrained(pretrained, prefix=‘’)
else:
checkpoint = torch.load(pretrained, map_location=“cpu”)
missing_keys, unexpected_keys = model.load_state_dict(checkpoint[“model”], strict=False)
print('Load pretrained model from: ’ + pretrained)

return model


def vit_base_patch16_224_ce(pretrained=False, **kwargs):
“”” ViT-Base model (ViT-B/16) from original paper (https://arxiv.org/abs/2010.11929).
“”“
model_kwargs = dict(
patch_size=16, embed_dim=768, depth=12, num_heads=12, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model


def vit_large_patch16_224_ce(pretrained=False, **kwargs):
“”” ViT-Large model (ViT-L/16) from original paper (https://arxiv.org/abs/2010.11929).
“”“
model_kwargs = dict(
patch_size=16, embed_dim=1024, depth=24, num_heads=16, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model中，对# 将 4输入分开，构建新的相同模态结合的2输入，2分支
import math
import logging
from functools import partial
from collections import OrderedDict
from copy import deepcopy

import torch
import torch.nn as nn
import torch.nn.functional as F

from timm.models.layers import to_2tuple

from lib.models.layers.patch_embed import PatchEmbed, PatchEmbed_event, xcorr_depthwise
from .utils import combine_tokens, recover_tokens
from .vit import VisionTransformer
from …layers.attn_blocks import CEBlock
from .new_counter_guide import Counter_Guide
from .ad_counter_guide import Counter_Guide_Enhanced
from …mamba_fetrack.models_mamba import VisionMamba, vim_small_patch16_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2
# 集成CrossMamba模块进行不同模态特征交互
from …mamba_fetrack.mamba_cross import CrossMamba
_logger = logging.getLogger(name)


class VisionTransformerCE(VisionTransformer):
“”” Vision Transformer with candidate elimination (CE) module

A PyTorch impl of : An Image is Worth 16x16 Words: Transformers for Image Recognition at Scale
- https://arxiv.org/abs/2010.11929

Includes distillation token & head support for DeiT: Data-efficient Image Transformers
- https://arxiv.org/abs/2012.12877
“”“
def init(self, img_size=224, patch_size=16, in_chans=3, num_classes=1000, embed_dim=768, depth=12,
num_heads=12, mlp_ratio=4., qkv_bias=True, representation_size=None, distilled=False,
drop_rate=0., attn_drop_rate=0., drop_path_rate=0., embed_layer=PatchEmbed, norm_layer=None,
act_layer=None, weight_init=‘’,
ce_loc=None, ce_keep_ratio=None):
super().init()
if isinstance(img_size, tuple):
self.img_size = img_size
else:
self.img_size = to_2tuple(img_size)
self.patch_size = patch_size
self.in_chans = in_chans

self.num_classes = num_classes
self.num_features = self.embed_dim = embed_dim # num_features for consistency with other models
self.num_tokens = 2 if distilled else 1
norm_layer = norm_layer or partial(nn.LayerNorm, eps=1e-6)
act_layer = act_layer or nn.GELU

self.patch_embed = embed_layer(
img_size=img_size, patch_size=patch_size, in_chans=in_chans, embed_dim=embed_dim)
num_patches = self.patch_embed.num_patches

self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim))
self.dist_token = nn.Parameter(torch.zeros(1, 1, embed_dim)) if distilled else None
self.pos_embed = nn.Parameter(torch.zeros(1, num_patches + self.num_tokens, embed_dim))
self.pos_drop = nn.Dropout(p=drop_rate)
self.pos_embed_event = PatchEmbed_event(in_chans=32, embed_dim=768, kernel_size=4, stride=4)
dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule
blocks = []
ce_index = 0
self.ce_loc = ce_loc
for i in range(depth):
ce_keep_ratio_i = 1.0
if ce_loc is not None and i in ce_loc:
ce_keep_ratio_i = ce_keep_ratio[ce_index]
ce_index += 1

blocks.append(
CEBlock(
dim=embed_dim, num_heads=num_heads, mlp_ratio=mlp_ratio, qkv_bias=qkv_bias, drop=drop_rate,
attn_drop=attn_drop_rate, drop_path=dpr[i], norm_layer=norm_layer, act_layer=act_layer,
keep_ratio_search=ce_keep_ratio_i)
)

self.blocks = nn.Sequential(*blocks)
self.norm = norm_layer(embed_dim)

self.init_weights(weight_init)

# 添加交互模块counter_guide
self.counter_guide = Counter_Guide_Enhanced(768, 768)
# 初始化CrossMamba模块
self.cross_mamba = CrossMamba(dim=768)
#########################################

def forward_features(self, z, x, event_z, event_x,
mask_z=None, mask_x=None,
ce_template_mask=None, ce_keep_rate=None,
return_last_attn=False
):
# 分支1 处理流程
B, H, W = x.shape[0], x.shape[2], x.shape[3]

x = self.patch_embed(x)
z = self.patch_embed(z)
z += self.pos_embed_z
x += self.pos_embed_x

if mask_z is not None and mask_x is not None:
mask_z = F.interpolate(mask_z[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_z = mask_z.flatten(1).unsqueeze(-1)

mask_x = F.interpolate(mask_x[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_x = mask_x.flatten(1).unsqueeze(-1)

mask_x = combine_tokens(mask_z, mask_x, mode=self.cat_mode)
mask_x = mask_x.squeeze(-1)

if self.add_cls_token:
cls_tokens = self.cls_token.expand(B, -1, -1)
cls_tokens = cls_tokens + self.cls_pos_embed
if self.add_sep_seg:
x += self.search_segment_pos_embed
z += self.template_segment_pos_embed
x = combine_tokens(z, x, mode=self.cat_mode)
if self.add_cls_token:
x = torch.cat([cls_tokens, x], dim=1)
x = self.pos_drop(x)

lens_z = self.pos_embed_z.shape[1]
lens_x = self.pos_embed_x.shape[1]
global_index_t = torch.linspace(0, lens_z - 1, lens_z).to(x.device)
global_index_t = global_index_t.repeat(B, 1)
global_index_s = torch.linspace(0, lens_x - 1, lens_x).to(x.device)
global_index_s = global_index_s.repeat(B, 1)
removed_indexes_s = []
# # 分支2 处理流程
event_x = self.pos_embed_event(event_x)
event_z = self.pos_embed_event(event_z)
event_x += self.pos_embed_x
event_z += self.pos_embed_z

event_x = combine_tokens(event_z, event_x, mode=self.cat_mode)

if self.add_cls_token:
event_x = torch.cat([cls_tokens, event_x], dim=1)

lens_z = self.pos_embed_z.shape[1]
lens_x = self.pos_embed_x.shape[1]

global_index_t1 = torch.linspace(0, lens_z - 1, lens_z).to(event_x.device)
global_index_t1 = global_index_t1.repeat(B, 1)
global_index_s1 = torch.linspace(0, lens_x - 1, lens_x).to(event_x.device)
global_index_s1 = global_index_s1.repeat(B, 1)

removed_indexes_s1 = []

for i, blk in enumerate(self.blocks):
# 第一个分支处理
x, global_index_t, global_index_s, removed_index_s, attn = <br/> blk(x, global_index_t, global_index_s, mask_x, ce_template_mask, ce_keep_rate)
# 第二个分支处理
event_x, global_index_t1, global_index_s1, removed_index_s1, attn1 = <br/> blk(event_x, global_index_t1, global_index_s1, mask_x, ce_template_mask, ce_keep_rate)


if self.ce_loc is not None and i in self.ce_loc:
removed_indexes_s.append(removed_index_s)
removed_indexes_s1.append(removed_index_s1)

# 在第1层增加counter_guide模块,验证早期融合效果
if i == 0 :
# 引入counter_guide进行特征增强
enhanced_x, enhanced_event_x = self.counter_guide(x, event_x)
# 将增强后的特征与原特征相加
x = x + enhanced_x
event_x = event_x + enhanced_event_x
##############
lens_x_new = global_index_s.shape[1]
lens_z_new = global_index_t.shape[1]
z = x[:, :lens_z_new]
x = x[:, lens_z_new:]
lens_eventx_new = global_index_s1.shape[1]
lens_eventz_new = global_index_t1.shape[1]
event_z = event_x[:, :lens_eventz_new]
event_x = event_x[:, lens_eventz_new:]
## 恢复原始的 token 顺序 #################################
if removed_indexes_s and removed_indexes_s[0] is not None:
removed_indexes_cat = torch.cat(removed_indexes_s, dim=1)
pruned_lens_x = lens_x - lens_x_new
pad_x = torch.zeros([B, pruned_lens_x, x.shape[2]], device=x.device)
x = torch.cat([x, pad_x], dim=1)
index_all = torch.cat([global_index_s, removed_indexes_cat], dim=1)
# 恢复原始顺序
C = x.shape[-1]
x = torch.zeros_like(x).scatter(dim=1, index=index_all.unsqueeze(-1).expand(B, -1, C).to(torch.int64), src=x)
x = recover_tokens(x, lens_z_new, lens_x, mode=self.cat_mode)

if removed_indexes_s1 and removed_indexes_s1[0] is not None:
removed_indexes_cat1 = torch.cat(removed_indexes_s1, dim=1)
pruned_lens_x1 = lens_x - lens_eventx_new
pad_x1 = torch.zeros([B, pruned_lens_x1, event_x.shape[2]], device=event_x.device)
event_x = torch.cat([event_x, pad_x1], dim=1)
index_all1 = torch.cat([global_index_s1, removed_indexes_cat1], dim=1)
# 恢复原始顺序
C1 = event_x.shape[-1]
event_x = torch.zeros_like(event_x).scatter(dim=1, index=index_all1.unsqueeze(-1).expand(B, -1, C1).to(torch.int64), src=event_x)
event_x = recover_tokens(event_x, lens_eventz_new, lens_x, mode=self.cat_mode)
################################################################

## 引入Crossmamba进行事件和图像特征融合,不采用concatenate
x = self.cross_mamba(x, event_x)
# 应用LayerNorm归一化处理
x = self.norm(x)

aux_dict = {
“attn”: attn,
‘attn1’: attn1,
“removed_indexes_s”: removed_indexes_s, # used for visualization
‘removed_indexes_s1’: removed_indexes_s1,
}

return x, aux_dict

def forward(self, z, x, event_z, event_x,
ce_template_mask=None, ce_keep_rate=None,
tnc_keep_rate=None,
return_last_attn=False):

x, aux_dict = self.forward_features(z, x, event_z, event_x, ce_template_mask=ce_template_mask, ce_keep_rate=ce_keep_rate,)

return x, aux_dict


def _create_vision_transformer(pretrained=False, **kwargs):
model = VisionTransformerCE(**kwargs)

if pretrained:
if ‘npz’ in pretrained:
model.load_pretrained(pretrained, prefix=‘’)
else:
checkpoint = torch.load(pretrained, map_location=“cpu”)
missing_keys, unexpected_keys = model.load_state_dict(checkpoint[“model”], strict=False)
print('Load pretrained model from: ’ + pretrained)

return model


def vit_base_patch16_224_ce(pretrained=False, **kwargs):
“”” ViT-Base model (ViT-B/16) from original paper (https://arxiv.org/abs/2010.11929).
“”“
model_kwargs = dict(
patch_size=16, embed_dim=768, depth=12, num_heads=12, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model


def vit_large_patch16_224_ce(pretrained=False, **kwargs):
“”” ViT-Large model (ViT-L/16) from original paper (https://arxiv.org/abs/2010.11929).
“”"
model_kwargs = dict(
patch_size=16, embed_dim=1024, depth=24, num_heads=16, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model中cross_mamba调用进行修改。直接给出x = self.cross_mamba(x, event_x)的修改，因为crossmamba的输入要求是3个。