| 247 | return out |
| 248 | |
| 249 | class ContrastiveBrainTextEncoder(nn.Module): |
| 250 | def __init__(self, pretrained_text_encoder, in_feature = 840, eeg_encoder_nhead=8, eeg_encoder_dim_feedforward = 2048, embed_dim = 768): |
| 251 | super(ContrastiveBrainTextEncoder, self).__init__() |
| 252 | # EEG Encoder |
| 253 | self.positional_embedding = PositionalEncoding(in_feature) |
| 254 | self.encoder_layer = nn.TransformerEncoderLayer(d_model=in_feature, nhead=eeg_encoder_nhead, dim_feedforward = eeg_encoder_dim_feedforward, batch_first=True) |
| 255 | self.EEG_Encoder = nn.TransformerEncoder(self.encoder_layer, num_layers=6) |
| 256 | self.EEG_pooler = Pooler(in_feature) |
| 257 | self.ln_final = nn.LayerNorm(in_feature) # to be considered |
| 258 | |
| 259 | # project to text embedding |
| 260 | self.EEG_projection = nn.Parameter(torch.empty(in_feature, embed_dim)) |
| 261 | |
| 262 | # Text Encoder |
| 263 | self.TextEncoder = pretrained_text_encoder |
| 264 | |
| 265 | # learned temperature parameter |
| 266 | self.logit_scale = nn.Parameter(torch.ones([]) * np.log(1 / 0.07)) |
| 267 | |
| 268 | def forward(self, input_EEG_features, input_EEG_attn_mask, input_ids, input_text_attention_masks): |
| 269 | # add positional embedding |
| 270 | input_EEG_features = self.positional_embedding(input_EEG_features) |
| 271 | # get EEG feature embedding |
| 272 | EEG_hiddenstates = self.EEG_Encoder(input_EEG_features, src_key_padding_mask = input_EEG_attn_mask) |
| 273 | EEG_hiddenstates = self.ln_final(EEG_hiddenstates) |
| 274 | EEG_features = self.EEG_pooler(EEG_hiddenstates) # [N, 840] |
| 275 | |
| 276 | # project to text embed size |
| 277 | EEG_features = EEG_features @ self.EEG_projection # [N, 768] |
| 278 | |
| 279 | # get text feature embedding |
| 280 | Text_features = self.TextEncoder(input_ids = input_ids, attention_mask = input_text_attention_masks, return_dict = True).pooler_output # [N, 768] |
| 281 | |
| 282 | # normalized features |
| 283 | EEG_features = EEG_features / EEG_features.norm(dim=-1, keepdim=True) # [N, 768] |
| 284 | Text_features = Text_features / Text_features.norm(dim=-1, keepdim=True) # [N, 768] |
| 285 | |
| 286 | # cosine similarity as logits |
| 287 | logit_scale = self.logit_scale.exp() |
| 288 | logits_per_EEG = logit_scale * EEG_features @ Text_features.t() # [N, N] |
| 289 | logits_per_text = logit_scale * Text_features @ EEG_features.t() # [N, N] |
| 290 | |
| 291 | return logits_per_EEG, logits_per_text |
nothing calls this directly
no outgoing calls
no test coverage detected