那么在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
import random
import numpy as np

_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)


def masking_fea(self,z, event_z, x, event_x, ratio=0.8):
b,nz,c = z.shape
b,nez,c = event_z.shape
b,nx,c = x.shape
b,nex,c = event_x.shape

assert(nz == nez)
assert(nx == nex)
lenz_out = int(nzratio)
lenx_out = int(nxratio)

mask_nz = torch.rand(b,nz).float()
mask_ez = torch.rand(b,nez).float()
mask_nx = torch.rand(b,nx).float()
mask_ex = torch.rand(b,nex).float()
mask_nz = mask_nz>0.4
mask_ez = mask_ez>0.4
mask_ez = ~mask_nz + mask_ez
mask_nz_idx = mask_nz.float().sort(1,descending=True)[-1].to(device = z.device)
mask_ez_idx = mask_ez.float().sort(1,descending=True)[-1].to(device = z.device)
mask_nx = mask_nx>0.4
mask_ex = mask_ex>0.4
mask_ex = ~mask_nx + mask_ex
mask_nx_idx = mask_nx.float().sort(1,descending=True)[-1].to(device = z.device)
mask_ex_idx = mask_ex.float().sort(1,descending=True)[-1].to(device = z.device)

masked_z = torch.gather(z, 1, mask_nz_idx[:,:lenz_out,None].repeat([1,1,c]))
masked_ez = torch.gather(event_z, 1, mask_ez_idx[:,:lenz_out,None].repeat([1,1,c]))
masked_x = torch.gather(x, 1, mask_nx_idx[:,:lenx_out,None].repeat([1,1,c]))
masked_ex = torch.gather(event_x, 1, mask_ex_idx[:,:lenx_out,None].repeat([1,1,c]))
return masked_z, masked_ez, masked_x, masked_ex,{‘x1’:mask_nx_idx[:,:lenx_out],‘x0’:mask_nx_idx[:,lenx_out:],
‘ex1’:mask_ex_idx[:,:lenx_out],‘ex0’:mask_ex_idx[:,lenx_out:], }

def masking_fea_rgb(self, z, x, ratio=0.8):
b, nz, c = z.shape
b, nx, c = x.shape

lenz_out = int(nz * ratio)
lenx_out = int(nx * ratio)

# 生成遮罩
mask_nz = torch.rand(b, nz).float() > 0.4
mask_nx = torch.rand(b, nx).float() > 0.4

# 获取遮罩后的索引
mask_nz_idx = mask_nz.float().sort(1, descending=True)[-1].to(device=z.device)
mask_nx_idx = mask_nx.float().sort(1, descending=True)[-1].to(device=x.device)

# 应用遮罩并且重构数据
masked_z = torch.gather(z, 1, mask_nz_idx[:, :lenz_out, None].repeat(1, 1, c))
masked_x = torch.gather(x, 1, mask_nx_idx[:, :lenx_out, None].repeat(1, 1, c))

return masked_z, masked_x, {‘z1’: mask_nz_idx[:, :lenz_out], ‘z0’: mask_nz_idx[:, lenz_out:],
‘x1’: mask_nx_idx[:, :lenx_out], ‘x0’: mask_nx_idx[:, lenx_out:]}

def masking_fea_event(self,event_z,event_x,ratio=0.8):
b,nez,c = event_z.shape
b,nex,c = event_x.shape
lenz_out = int(nezratio)
lenx_out = int(nex*ratio)
mask_ez = torch.rand(b,nez).float() > 0.4
mask_ex = torch.rand(b,nex).float() > 0.4
mask_ez_idx = mask_ez.float().sort(1,descending=True)[-1].to(device = event_z.device)
mask_ex_idx = mask_ex.float().sort(1, descending=True)[-1].to(device=event_x.device)
masked_ez = torch.gather(event_z, 1, mask_ez_idx[:,:lenz_out,None].repeat([1,1,c]))
masked_ex = torch.gather(event_x, 1, mask_ex_idx[:,:lenx_out,None].repeat([1,1,c]))

return masked_ez, masked_ex , {‘ez1’: mask_ez_idx[:, :lenz_out], ‘ez0’: mask_ez_idx[:, lenz_out:],
‘ex1’: mask_ex_idx[:, :lenx_out], ‘ex0’: mask_ex_idx[:, lenx_out:]}

############ 针对MHA中集成masking_fea重构masking操作 ############################
def masking_fea_combined(self, x_combined, event_x_combined, ratio=0.8):
b, n, c = x_combined.shape
b_event, n_event, c_event = event_x_combined.shape

assert n == n_event # 确保合并后的数据维度匹配

len_out = int(n * ratio)

mask = torch.rand(b, n).float()
mask_event = torch.rand(b_event, n_event).float()
mask = mask > 0.4
mask_event = mask_event > 0.4

#### 此处新增 不同模态互补性
mask_event = ~mask + mask_event

mask_idx = mask.float().sort(1, descending=True)[-1].to(device=x_combined.device)
mask_event_idx = mask_event.float().sort(1, descending=True)[-1].to(device=event_x_combined.device)

masked_x = torch.gather(x_combined, 1, mask_idx[:, :len_out, None].repeat(1, 1, c))
masked_event_x = torch.gather(event_x_combined, 1, mask_event_idx[:, :len_out, None].repeat(1, 1, c_event))

return masked_x, masked_event_x, {‘x1’: mask_idx[:, :len_out],
‘x0’: mask_idx[:, len_out:],
‘ex1’: mask_event_idx[:, :len_out],
‘ex0’: mask_event_idx[:, len_out:]
}
################################################################################################



def reorder(x, idx1, idx0):
‘’‘Reorder function using the indices provided by masking_fea.’‘’
B,N, C = x.shape
combined_idx = torch.cat([idx1, idx0], dim=1)
x_reordered = torch.zeros_like(x)
x_reordered = x_reordered.scatter(1, combined_idx.unsqueeze(-1).expand(-1, -1, C), x)
return x_reordered
# masking之后的填充和重排序
def pad_reorder(x, idx1, idx0):
‘’‘Reorder function using the indices provided by masking_fea.’‘’
B,N,C = x.shape
pad_x = torch.zeros([B, idx0.shape[1], C], device=x.device)
x = torch.cat([x, pad_x], dim=1)
combined_idx = torch.cat([idx1, idx0], dim=1)
x_reordered = torch.zeros_like(x).scatter(1, combined_idx.unsqueeze(-1).expand(B, -1, C).to(torch.int64), x)
return x_reordered



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

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

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 = 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):
######################## 每隔一层应用遮罩处理 ###########################
if i % 2 == 0:
x, event_x, token_idx = self.masking_fea_combined(x,event_x,ratio=0.95)
# 重排序逻辑
# x = self.reorder(x, token_idx[‘x1’], token_idx[‘x0’])
# event_x = self.reorder(event_x, token_idx[‘ex1’], token_idx[‘ex0’])
########################################################################
# 第一个分支处理
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)

# 为了不影响masking_fea的随机掩膜操作，在此处增加 填充 & 重排序 逻辑
x = self.pad_reorder(x, token_idx[‘x1’], token_idx[‘x0’])
event_x = self.pad_reorder(event_x, token_idx[‘ex1’], token_idx[‘ex0’])

# 应用LayerNorm归一化处理
x = self.norm(x)
event_x = self.norm(event_x)

x_cat = torch.cat([event_x, x], dim=1)
x = x_cat # cat仅仅是做 rgb 和 event voxel拼接，数据增强采用的随机遮蔽特征策略。


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
中，此处
for i, blk in enumerate(self.blocks):
######################## 每隔一层应用遮罩处理 ###########################
if i % 2 == 0:
x, event_x, token_idx = self.masking_fea_combined(x,event_x,ratio=0.95)
# 重排序逻辑
# x = self.reorder(x, token_idx[‘x1’], token_idx[‘x0’])
# event_x = self.reorder(event_x, token_idx[‘ex1’], token_idx[‘ex0’])
########################################################################
# 第一个分支处理
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)

# 为了不影响masking_fea的随机掩膜操作，在此处增加 填充 & 重排序 逻辑
x = self.pad_reorder(x, token_idx[‘x1’], token_idx[‘x0’])
event_x = self.pad_reorder(event_x, token_idx[‘ex1’], token_idx[‘ex0’])

# 应用LayerNorm归一化处理
x = self.norm(x)
event_x = self.norm(event_x)

x_cat = torch.cat([event_x, x], dim=1)
x = x_cat 是否需要引入recover_tokens？